diff --git a/Cargo.lock b/Cargo.lock index 7fa8a031..64ad2d98 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1401,6 +1401,16 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "slab" version = "0.4.12" @@ -1538,6 +1548,7 @@ dependencies = [ "libc", "mio", "pin-project-lite", + "signal-hook-registry", "socket2", "tokio-macros", "windows-sys 0.61.2", diff --git a/Cargo.toml b/Cargo.toml index 896d376c..f7a34ef5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -67,7 +67,7 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls" rusqlite = { version = "0.32", default-features = false, features = ["bundled", "hooks", "limits"], optional = true } url = { version = "2", optional = true } futures-util = { version = "0.3", optional = true } -tokio = { version = "1", features = ["rt-multi-thread", "net", "time", "sync"], optional = true } +tokio = { version = "1", features = ["rt-multi-thread", "net", "time", "sync", "fs", "io-util", "process"], optional = true } edge_abi = { package = "pd-edge-abi", version = "0.1.1", default-features = false, optional = true } futures-channel = "0.3" paste = "1" diff --git a/build.rs b/build.rs index f5c1deb2..ce572c4f 100644 --- a/build.rs +++ b/build.rs @@ -251,10 +251,21 @@ fn write_generated_file(path: &Path, contents: &str) { fn builtin_source_specs(namespaces: &[NamespaceDecl]) -> Vec { namespaces .iter() - .map(|namespace| SourceSpec { - path: format!("src/builtins/runtime/{}.rs", namespace.module), - module: namespace.module.clone(), - category: SourceCategory::NamespacedBuiltin, + .map(|namespace| { + let path = if namespace.module == "io" { + if cfg!(feature = "async") { + "src/builtins/runtime/io/async_io.rs".to_string() + } else { + "src/builtins/runtime/io/blocking.rs".to_string() + } + } else { + format!("src/builtins/runtime/{}.rs", namespace.module) + }; + SourceSpec { + path, + module: namespace.module.clone(), + category: SourceCategory::NamespacedBuiltin, + } }) .collect() } @@ -2108,7 +2119,7 @@ fn type_label(ty: &Type) -> String { }; format!("{} | null", type_label(inner)) } - "VmResult" | "HostCallResult" => { + "VmResult" | "HostCallResult" | "HostFutureOutput" => { let syn::PathArguments::AngleBracketed(args) = &segment.arguments else { panic!("{ident} requires one generic argument"); }; diff --git a/crates/rustscript/tests/alias_smoke.rs b/crates/rustscript/tests/alias_smoke.rs index 1a63ee26..af2b24a8 100644 --- a/crates/rustscript/tests/alias_smoke.rs +++ b/crates/rustscript/tests/alias_smoke.rs @@ -1,3 +1,6 @@ +#[cfg(feature = "sqlite")] +use rustscript::SqliteHostExt; + /// Verify that the `rustscript` alias crate re-exports the same API as `pd-vm`. #[test] fn alias_exports_compile_source() { diff --git a/pd-host-function/src/lib.rs b/pd-host-function/src/lib.rs index a6b35147..fb4f8f96 100644 --- a/pd-host-function/src/lib.rs +++ b/pd-host-function/src/lib.rs @@ -308,13 +308,13 @@ fn generate_vm_wrapper( Ok(quote! { #[allow(dead_code)] - pub(super) fn #wrapper_name(#(#imm_wrapper_params),*) -> #wrapper_output { + pub(crate) fn #wrapper_name(#(#imm_wrapper_params),*) -> #wrapper_output { #(#imm_extract_stmts)* #call_expr } #[allow(dead_code)] - pub(super) fn #mutable_wrapper_name(#(#mut_wrapper_params),*) -> #wrapper_output { + pub(crate) fn #mutable_wrapper_name(#(#mut_wrapper_params),*) -> #wrapper_output { #(#mut_extract_stmts)* #call_expr } @@ -344,7 +344,7 @@ fn generate_async_vm_wrapper( let ty = &pat_type.ty; if is_host_context_param(input) { extract_stmts.push(quote! { - let #ident = <#ty as super::CaptureAsyncHostContext>::capture(vm)?; + let #ident = <#ty as super::CaptureAsyncHostContext>::capture_with_args(vm, args)?; }); call_args.push(quote!(#ident)); continue; @@ -366,12 +366,14 @@ fn generate_async_vm_wrapper( } else { quote!(#impl_name(#(#call_args),*).await) }; - let body = quote! { - #(#extract_stmts)* - vm.submit_host_future(Box::pin(async move { - let value = #await_value; + let future_result = if return_is_host_future_output(&item.sig.output) { + quote!(Ok(value.map(super::return_one))) + } else { + quote! { match super::IntoHostCallOutcome::into_host_call_outcome(value) { - super::CallOutcome::Return(values) => Ok(values), + super::CallOutcome::Return(values) => { + Ok(super::HostFutureOutput::returning(values)) + } super::CallOutcome::Pending(op_id) => Err(super::VmError::HostError( format!("async host function returned nested pending operation {op_id}"), )), @@ -381,12 +383,19 @@ fn generate_async_vm_wrapper( ), ), } + } + }; + let body = quote! { + #(#extract_stmts)* + vm.submit_host_future(Box::pin(async move { + let value = #await_value; + #future_result })) }; Ok(quote! { #[allow(dead_code)] - pub(super) fn #wrapper_name( + pub(crate) fn #wrapper_name( vm: &mut super::super::Vm, args: &[super::super::Value], ) -> super::super::VmResult { @@ -394,7 +403,7 @@ fn generate_async_vm_wrapper( } #[allow(dead_code)] - pub(super) fn #mutable_wrapper_name( + pub(crate) fn #mutable_wrapper_name( vm: &mut super::super::Vm, args: &mut [super::super::Value], ) -> super::super::VmResult { @@ -465,6 +474,20 @@ fn unwrap_vm_result_type(ty: &Type) -> Result, Error> { } } +fn return_is_host_future_output(output: &ReturnType) -> bool { + vm_result_inner_type(output) + .expect("pd_host_function return type should already be validated") + .and_then(|ty| match ty { + Type::Path(path) => path + .path + .segments + .last() + .map(|segment| segment.ident.clone()), + _ => None, + }) + .is_some_and(|ident| ident == "HostFutureOutput") +} + fn return_is_vm_result(output: &ReturnType) -> bool { vm_result_inner_type(output) .expect("pd_host_function return type should already be validated") @@ -526,7 +549,7 @@ fn type_label(ty: &Type) -> Result { let inner_label = type_label(inner)?; Ok(format!("{inner_label} | null")) } - "VmResult" | "HostCallResult" => { + "VmResult" | "HostCallResult" | "HostFutureOutput" => { let syn::PathArguments::AngleBracketed(args) = &segment.arguments else { return Err(Error::new_spanned( &segment.arguments, @@ -720,9 +743,26 @@ mod tests { assert!(expanded.contains("async move")); assert!(expanded.contains("borrow_arg")); assert!(expanded.contains("CaptureAsyncHostContext")); + assert!(expanded.contains("capture_with_args")); assert!(!expanded.contains("pd_host_context")); } + #[test] + fn async_host_future_output_maps_its_inner_value_to_call_return() { + let attr: Punctuated = parse_quote!(name = "test::completion"); + let item: ItemFn = parse_quote! { + /// Completes after mutating VM-owned state. + async fn completion() -> VmResult> { + todo!() + } + }; + + let expanded = expand_pd_host_function(attr, item) + .expect("host future output should be accepted") + .to_string(); + assert!(expanded.contains("value . map (super :: return_one)")); + } + #[test] fn async_signature_rejects_borrowed_parameters() { let attr: Punctuated = parse_quote!(name = "test::borrowed"); diff --git a/src/builtins/runtime/cancellation.rs b/src/builtins/runtime/cancellation.rs index ef8dbc67..85b53c16 100644 --- a/src/builtins/runtime/cancellation.rs +++ b/src/builtins/runtime/cancellation.rs @@ -410,6 +410,7 @@ impl OperationState { self.core.status() } + #[cfg_attr(feature = "async", allow(dead_code))] pub fn set_payload(&self, payload: ResourceHandle) { self.core .inner @@ -453,6 +454,7 @@ impl OperationState { .payload } + #[cfg_attr(feature = "async", allow(dead_code))] pub fn set_resource(&self, resource: ResourceHandle) { self.core .inner @@ -560,6 +562,7 @@ impl OperationRegistry { Ok(id) } + #[cfg_attr(feature = "async", allow(dead_code))] pub fn start_owned( &mut self, owner: OperationOwner, diff --git a/src/builtins/runtime/http.rs b/src/builtins/runtime/http.rs index 9f11e891..bdb4b3fb 100644 --- a/src/builtins/runtime/http.rs +++ b/src/builtins/runtime/http.rs @@ -1,17 +1,22 @@ #[cfg(feature = "async")] use futures_util::StreamExt; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +#[cfg(feature = "async")] +use std::sync::atomic::Ordering; #[cfg(feature = "async")] use pd_host_function::pd_host_function; #[cfg(feature = "async")] -use super::{Vm, VmMap, VmResult}; +use super::{VmMap, VmResult}; #[cfg(feature = "async")] use crate::builtins::runtime::cancellation::{CancellationReason, CancellationToken}; #[cfg(feature = "async")] use crate::vm::CaptureAsyncHostContext; #[cfg(feature = "async")] use crate::vm::Value; +use crate::vm::Vm; #[cfg(feature = "async")] use crate::vm::VmError; @@ -44,50 +49,86 @@ impl Default for HttpConfig { } } -pub(crate) struct HttpState { +#[derive(Default)] +struct HttpHostState { #[cfg(feature = "async")] config: Option, - pub(crate) max_in_flight: usize, + max_in_flight: usize, + in_flight: Arc, } -impl Default for HttpState { - fn default() -> Self { - Self { - #[cfg(feature = "async")] - config: None, - max_in_flight: crate::builtins::runtime::cancellation::DEFAULT_MAX_PENDING_OPERATIONS, - } - } +/// HTTP host configuration owned by the HTTP host implementation. +pub trait HttpHostExt { + fn configure_http(&mut self, config: HttpConfig); + fn set_http_max_in_flight(&mut self, max_in_flight: usize); + fn http_max_in_flight(&self) -> usize; + fn clear_http_configuration(&mut self); + fn http_is_configured(&self) -> bool; } -impl HttpState { - pub(crate) fn reset_for_reuse(&mut self) {} - - pub(crate) fn configure(&mut self, config: HttpConfig) { - #[cfg(feature = "async")] - { - self.config = Some(config); - } +impl HttpHostExt for Vm { + fn configure_http(&mut self, config: HttpConfig) { + let (max_in_flight, in_flight) = self + .host + .host_function_state::() + .map_or_else( + || { + ( + crate::builtins::runtime::cancellation::DEFAULT_MAX_PENDING_OPERATIONS, + Arc::new(AtomicUsize::new(0)), + ) + }, + |state| (state.max_in_flight, Arc::clone(&state.in_flight)), + ); + self.host.set_host_function_state(HttpHostState { + #[cfg(feature = "async")] + config: Some(config), + max_in_flight, + in_flight, + }); #[cfg(not(feature = "async"))] let _ = config; } - pub(crate) fn clear_configuration(&mut self) { - #[cfg(feature = "async")] - { - self.config = None; + fn set_http_max_in_flight(&mut self, max_in_flight: usize) { + if self.host.host_function_state::().is_none() { + self.host.set_host_function_state(HttpHostState { + #[cfg(feature = "async")] + config: None, + max_in_flight: + crate::builtins::runtime::cancellation::DEFAULT_MAX_PENDING_OPERATIONS, + in_flight: Arc::new(AtomicUsize::new(0)), + }); } + self.host + .host_function_state_mut::() + .expect("HTTP host state was inserted") + .max_in_flight = max_in_flight; } - #[cfg(all(test, feature = "async"))] - pub(crate) fn configuration(&self) -> Option<&HttpConfig> { - self.config.as_ref() + fn http_max_in_flight(&self) -> usize { + self.host.host_function_state::().map_or( + crate::builtins::runtime::cancellation::DEFAULT_MAX_PENDING_OPERATIONS, + |state| state.max_in_flight, + ) } - pub(crate) fn is_configured(&self) -> bool { + fn clear_http_configuration(&mut self) { + crate::builtins::runtime::cancel_operations_by_owner( + self, + crate::builtins::runtime::cancellation::OperationOwner::Http, + crate::builtins::runtime::cancellation::CancellationReason::Requested, + ); + self.host.remove_host_function_state::(); + } + + fn http_is_configured(&self) -> bool { #[cfg(feature = "async")] { - self.config.is_some() + self.host + .host_function_state::() + .and_then(|state| state.config.as_ref()) + .is_some() } #[cfg(not(feature = "async"))] false @@ -111,20 +152,65 @@ fn cancellation_vm_error(token: &CancellationToken) -> VmError { pub(super) struct HttpRequestContext { config: HttpConfig, cancellation: CancellationToken, + _permit: HttpInFlightPermit, +} + +#[cfg(feature = "async")] +struct HttpInFlightPermit { + active: Arc, +} + +#[cfg(feature = "async")] +impl HttpInFlightPermit { + fn acquire(state: &HttpHostState) -> VmResult { + let mut active = state.in_flight.load(Ordering::Acquire); + loop { + if active >= state.max_in_flight { + return Err(VmError::HostError(format!( + "HTTP in-flight request limit of {} was reached", + state.max_in_flight + ))); + } + match state.in_flight.compare_exchange_weak( + active, + active + 1, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => { + return Ok(Self { + active: Arc::clone(&state.in_flight), + }); + } + Err(observed) => active = observed, + } + } + } +} + +#[cfg(feature = "async")] +impl Drop for HttpInFlightPermit { + fn drop(&mut self) { + self.active.fetch_sub(1, Ordering::AcqRel); + } } #[cfg(feature = "async")] impl CaptureAsyncHostContext for HttpRequestContext { fn capture(vm: &mut Vm) -> VmResult { - let config = vm + let state = vm .host - .http_state + .host_function_state::() + .ok_or_else(|| VmError::HostError("HTTP host is not configured".to_string()))?; + let config = state .config .clone() .ok_or_else(|| VmError::HostError("HTTP host is not configured".to_string()))?; + let permit = HttpInFlightPermit::acquire(state)?; Ok(Self { config, cancellation: CancellationToken::root(), + _permit: permit, }) } } @@ -545,6 +631,8 @@ async fn execute_request( mod tests { use super::HttpConfig; #[cfg(feature = "async")] + use super::HttpHostExt; + #[cfg(feature = "async")] use super::{ CancellationReason, HttpRequest, VmMap, builtin_http_client_request, execute_request, is_restricted_ip, validate_resolved_addresses, validate_url, diff --git a/src/builtins/runtime/io/async_io.rs b/src/builtins/runtime/io/async_io.rs new file mode 100644 index 00000000..d38bf818 --- /dev/null +++ b/src/builtins/runtime/io/async_io.rs @@ -0,0 +1,580 @@ +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::Duration; + +#[cfg(unix)] +use std::os::unix::process::CommandExt; + +use pd_host_function::pd_host_function; +use tokio::fs::{File, OpenOptions}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; +use tokio::process::{Child, ChildStdin, ChildStdout, Command}; +use tokio::sync::Mutex; + +use super::super::resource::ResourceTypeId; +use super::super::{ + CancellationReason, CaptureAsyncHostContext, HostFutureOutput, HostOpId, ResourceHandle, + RuntimeError, RuntimeErrorCode, Value, Vm, VmError, VmResult, +}; +use super::{IoPolicy, io_policy}; + +#[derive(Debug)] +pub(crate) enum IoHandle { + File(BufReader), + PopenRead { + child: Child, + stdout: BufReader, + }, + PopenWrite { + child: Child, + stdin: ChildStdin, + }, +} + +struct IoResource { + handle: Mutex>, + process_id: AtomicU32, +} + +impl IoResource { + fn new(handle: IoHandle) -> Self { + let process_id = match &handle { + IoHandle::PopenRead { child, .. } | IoHandle::PopenWrite { child, .. } => { + child.id().unwrap_or(0) + } + IoHandle::File(_) => 0, + }; + Self { + handle: Mutex::new(Some(handle)), + process_id: AtomicU32::new(process_id), + } + } + + async fn take_handle(&self) -> VmResult { + self.handle + .lock() + .await + .take() + .ok_or_else(|| VmError::HostError("io handle is closed".to_string())) + } + + fn close(&self, reason: CancellationReason) -> VmResult<()> { + if let Ok(mut handle) = self.handle.try_lock() + && let Some(handle) = handle.take() + { + start_close_io_handle(handle, reason)?; + } + terminate_process_id(self.process_id.load(Ordering::Acquire), reason)?; + self.process_id.store(0, Ordering::Release); + Ok(()) + } +} + +impl Drop for IoResource { + fn drop(&mut self) { + if let Some(handle) = self.handle.get_mut().take() { + let _ = start_close_io_handle(handle, CancellationReason::VmReset); + } + let _ = terminate_process_id( + self.process_id.load(Ordering::Acquire), + CancellationReason::VmReset, + ); + } +} + +#[derive(Clone)] +pub(crate) struct IoPolicyContext { + policy: Option, +} + +impl CaptureAsyncHostContext for IoPolicyContext { + fn capture(vm: &mut Vm) -> VmResult { + Ok(Self { + policy: io_policy(vm), + }) + } +} + +pub(crate) struct IoHandleContext { + handle: ResourceHandle, + resource: Arc, + max_read_bytes: Option, + max_write_bytes: Option, +} + +impl CaptureAsyncHostContext for IoHandleContext { + fn capture(_vm: &mut Vm) -> VmResult { + Err(VmError::HostError( + "io handle context requires call arguments".to_string(), + )) + } + + fn capture_with_args(vm: &mut Vm, args: &[Value]) -> VmResult { + let handle_id = match args.first() { + Some(Value::Int(value)) => *value, + Some(_) => return Err(VmError::TypeMismatch("int")), + None => return Err(VmError::HostError("missing io handle argument".to_string())), + }; + let handle = resource_handle(handle_id)?; + let resource = io_resource_for_handle(vm, handle)?; + Ok(Self { + handle, + resource, + max_read_bytes: io_policy(vm).map(|policy| policy.max_read_bytes), + max_write_bytes: io_policy(vm).map(|policy| policy.max_write_bytes), + }) + } +} + +/// Opens a file handle for runtime I/O. +#[pd_host_function(name = "io::open")] +pub(crate) async fn builtin_io_open( + #[pd_host_context] context: IoPolicyContext, + path: String, + mode: String, +) -> VmResult> { + let writes = match mode.as_str() { + "r" => false, + "w" | "a" | "r+" | "w+" | "a+" => true, + other => { + return Err(VmError::HostError(format!( + "io_open unsupported mode '{other}'" + ))); + } + }; + let path = authorize_io_path(context.policy.as_ref(), &path, writes).await?; + let mut options = OpenOptions::new(); + match mode.as_str() { + "r" => { + options.read(true); + } + "w" => { + options.write(true).create(true).truncate(true); + } + "a" => { + options.append(true).create(true); + } + "r+" => { + options.read(true).write(true); + } + "w+" => { + options.read(true).write(true).create(true).truncate(true); + } + "a+" => { + options.read(true).append(true).create(true); + } + _ => unreachable!(), + } + let file = options + .open(path) + .await + .map_err(|error| VmError::HostError(format!("io_open failed: {error}")))?; + let handle = IoHandle::File(BufReader::new(file)); + Ok(HostFutureOutput::complete(move |vm| { + let handle = insert_io_resource(vm, handle)?; + match handle.as_value() { + Value::Int(value) => Ok(value), + _ => unreachable!(), + } + })) +} + +/// Starts a child process and returns a process-backed handle. +#[pd_host_function(name = "io::popen")] +pub(crate) async fn builtin_io_popen( + #[pd_host_context] context: IoPolicyContext, + command: String, + mode: String, +) -> VmResult> { + if mode != "r" && mode != "w" { + return Err(VmError::HostError(format!( + "io_popen unsupported mode '{mode}'" + ))); + } + if !context + .policy + .as_ref() + .is_none_or(|policy| policy.allow_process) + { + return Err(VmError::HostError( + "io_popen requires the command capability".to_string(), + )); + } + let handle = spawn_shell_command(&command, &mode)?; + Ok(HostFutureOutput::complete(move |vm| { + let handle = insert_io_resource(vm, handle)?; + match handle.as_value() { + Value::Int(value) => Ok(value), + _ => unreachable!(), + } + })) +} + +/// Reads all remaining text from an I/O handle. +#[pd_host_function(name = "io::read_all")] +pub(crate) async fn builtin_io_read_all( + #[pd_host_context] context: IoHandleContext, + _handle_id: i64, +) -> VmResult> { + let mut guard = context.resource.handle.lock().await; + let handle = guard + .as_mut() + .ok_or_else(|| VmError::HostError("io handle is closed".to_string()))?; + let mut out = String::new(); + match handle { + IoHandle::File(file) => file.read_to_string(&mut out).await, + IoHandle::PopenRead { stdout, .. } => stdout.read_to_string(&mut out).await, + IoHandle::PopenWrite { .. } => { + return Err(VmError::HostError( + "io_read_all cannot read from a write handle".to_string(), + )); + } + } + .map_err(|error| VmError::HostError(format!("io_read_all failed: {error}")))?; + if context + .max_read_bytes + .is_some_and(|limit| out.len() > limit) + { + return Err(VmError::HostError( + "io_read_all exceeded read limit".to_string(), + )); + } + Ok(HostFutureOutput::returning(out)) +} + +/// Reads a single line of text from an I/O handle. +#[pd_host_function(name = "io::read_line")] +pub(crate) async fn builtin_io_read_line( + #[pd_host_context] context: IoHandleContext, + _handle_id: i64, +) -> VmResult> { + let mut guard = context.resource.handle.lock().await; + let handle = guard + .as_mut() + .ok_or_else(|| VmError::HostError("io handle is closed".to_string()))?; + let mut line = String::new(); + match handle { + IoHandle::File(file) => file.read_line(&mut line).await, + IoHandle::PopenRead { stdout, .. } => stdout.read_line(&mut line).await, + IoHandle::PopenWrite { .. } => { + return Err(VmError::HostError( + "io_read_line cannot read from a write handle".to_string(), + )); + } + } + .map_err(|error| VmError::HostError(format!("io_read_line failed: {error}")))?; + if context + .max_read_bytes + .is_some_and(|limit| line.len() > limit) + { + return Err(VmError::HostError( + "io_read_line exceeded read limit".to_string(), + )); + } + Ok(HostFutureOutput::returning(line)) +} + +/// Writes text to an I/O handle. +#[pd_host_function(name = "io::write")] +pub(crate) async fn builtin_io_write( + #[pd_host_context] context: IoHandleContext, + _handle_id: i64, + text: String, +) -> VmResult> { + if context + .max_write_bytes + .is_some_and(|limit| text.len() > limit) + { + return Err(VmError::HostError( + "io_write exceeded write limit".to_string(), + )); + } + let mut guard = context.resource.handle.lock().await; + let handle = guard + .as_mut() + .ok_or_else(|| VmError::HostError("io handle is closed".to_string()))?; + let written = match handle { + IoHandle::File(file) => file.get_mut().write(text.as_bytes()).await, + IoHandle::PopenWrite { stdin, .. } => stdin.write(text.as_bytes()).await, + IoHandle::PopenRead { .. } => { + return Err(VmError::HostError( + "io_write cannot write to a read handle".to_string(), + )); + } + } + .map_err(|error| VmError::HostError(format!("io_write failed: {error}")))?; + Ok(HostFutureOutput::returning(written as i64)) +} + +/// Flushes buffered output for an I/O handle. +#[pd_host_function(name = "io::flush")] +pub(crate) async fn builtin_io_flush( + #[pd_host_context] context: IoHandleContext, + _handle_id: i64, +) -> VmResult> { + let mut guard = context.resource.handle.lock().await; + let handle = guard + .as_mut() + .ok_or_else(|| VmError::HostError("io handle is closed".to_string()))?; + match handle { + IoHandle::File(file) => file.get_mut().flush().await, + IoHandle::PopenWrite { stdin, .. } => stdin.flush().await, + IoHandle::PopenRead { .. } => Ok(()), + } + .map_err(|error| VmError::HostError(format!("io_flush failed: {error}")))?; + Ok(HostFutureOutput::returning(true)) +} + +/// Closes an I/O handle. +#[pd_host_function(name = "io::close")] +pub(crate) async fn builtin_io_close( + #[pd_host_context] context: IoHandleContext, + _handle_id: i64, +) -> VmResult> { + let resource = context.resource; + let handle = context.handle; + let resource_handle = resource.take_handle().await?; + let close_result = close_io_handle(resource_handle, CancellationReason::ResourceClosed).await; + Ok(HostFutureOutput::complete(move |vm| { + super::super::close_runtime_resource(vm, handle, CancellationReason::ResourceClosed) + .map_err(runtime_host_error)?; + close_result?; + Ok(true) + })) +} + +/// Returns whether a file system path exists. +#[pd_host_function(name = "io::exists")] +pub(crate) async fn builtin_io_exists( + #[pd_host_context] context: IoPolicyContext, + path: String, +) -> VmResult> { + let path = authorize_io_path(context.policy.as_ref(), &path, false).await?; + let exists = tokio::fs::try_exists(path) + .await + .map_err(|error| VmError::HostError(format!("io_exists failed: {error}")))?; + Ok(HostFutureOutput::returning(exists)) +} + +#[allow(dead_code)] +pub(crate) fn cancel_builtin_io_op_with_reason( + _vm: &mut Vm, + _op_id: HostOpId, + _reason: CancellationReason, +) { +} + +async fn authorize_io_path( + policy: Option<&IoPolicy>, + path: &str, + writes: bool, +) -> VmResult { + let requested = PathBuf::from(path); + let Some(policy) = policy else { + return Ok(requested); + }; + if writes && !policy.allow_write { + return Err(VmError::HostError( + "io path write requires the write capability".to_string(), + )); + } + let absolute = if requested.is_absolute() { + requested + } else { + std::env::current_dir() + .map_err(|error| VmError::HostError(format!("io path resolution failed: {error}")))? + .join(requested) + }; + let canonical = canonicalize_io_target(&absolute).await?; + for root in &policy.allowed_roots { + let root = tokio::fs::canonicalize(Path::new(root)) + .await + .map_err(|error| { + VmError::HostError(format!( + "io allowed root '{root}' cannot be resolved: {error}" + )) + })?; + if canonical.starts_with(root) { + return Ok(canonical); + } + } + Err(VmError::HostError(format!( + "io path '{}' is outside the allowed roots", + canonical.display() + ))) +} + +async fn canonicalize_io_target(path: &Path) -> VmResult { + if tokio::fs::try_exists(path) + .await + .map_err(|error| VmError::HostError(format!("io path resolution failed: {error}")))? + { + return tokio::fs::canonicalize(path) + .await + .map_err(|error| VmError::HostError(format!("io path resolution failed: {error}"))); + } + let parent = path + .parent() + .ok_or_else(|| VmError::HostError(format!("io path '{}' has no parent", path.display())))?; + let file_name = path.file_name().ok_or_else(|| { + VmError::HostError(format!("io path '{}' has no file name", path.display())) + })?; + tokio::fs::canonicalize(parent) + .await + .map(|parent| parent.join(file_name)) + .map_err(|error| VmError::HostError(format!("io path resolution failed: {error}"))) +} + +fn spawn_shell_command(command: &str, mode: &str) -> VmResult { + let mut process = if cfg!(windows) { + let mut cmd = Command::new("cmd"); + cmd.arg("/C").arg(command); + cmd + } else { + let mut cmd = Command::new("sh"); + cmd.arg("-c").arg(command); + cmd + }; + #[cfg(unix)] + process.as_std_mut().process_group(0); + process.kill_on_drop(true); + match mode { + "r" => { + process.stdout(Stdio::piped()).stdin(Stdio::null()); + } + "w" => { + process.stdin(Stdio::piped()).stdout(Stdio::null()); + } + _ => {} + } + let mut child = process + .spawn() + .map_err(|error| VmError::HostError(format!("io_popen failed: {error}")))?; + match mode { + "r" => { + let stdout = child.stdout.take().ok_or_else(|| { + VmError::HostError("io_popen failed to capture stdout".to_string()) + })?; + Ok(IoHandle::PopenRead { + child, + stdout: BufReader::new(stdout), + }) + } + "w" => { + let stdin = child.stdin.take().ok_or_else(|| { + VmError::HostError("io_popen failed to capture stdin".to_string()) + })?; + Ok(IoHandle::PopenWrite { child, stdin }) + } + _ => unreachable!(), + } +} + +fn resource_handle(handle_id: i64) -> VmResult { + if handle_id <= 0 { + return Err(VmError::HostError(format!( + "invalid io handle id {handle_id}; expected positive handle id" + ))); + } + ResourceHandle::from_value(&Value::Int(handle_id)).map_err(runtime_host_error) +} + +fn io_resource_for_handle(vm: &Vm, handle: ResourceHandle) -> VmResult> { + vm.host + .runtime_resources + .get::>(handle, ResourceTypeId::IO_FILE) + .cloned() + .map_err(runtime_host_error) +} + +fn insert_io_resource(vm: &mut Vm, handle: IoHandle) -> VmResult { + vm.host + .runtime_resources + .insert_with_cleanup( + ResourceTypeId::IO_FILE, + Arc::new(IoResource::new(handle)), + |resource, reason| resource.close(reason).map_err(io_cleanup_error), + ) + .map_err(runtime_host_error) +} + +fn runtime_host_error(error: impl std::fmt::Display) -> VmError { + VmError::HostError(error.to_string()) +} + +fn io_cleanup_error(error: VmError) -> RuntimeError { + RuntimeError::new( + RuntimeErrorCode::ResourceCleanupFailed, + "io::close", + error.to_string(), + ) +} + +async fn close_io_handle(mut handle: IoHandle, reason: CancellationReason) -> VmResult<()> { + match &mut handle { + IoHandle::File(file) => { + file.get_mut() + .flush() + .await + .map_err(|error| VmError::HostError(format!("io close failed: {error}")))?; + } + IoHandle::PopenRead { child, .. } => wait_for_child(child, reason).await?, + IoHandle::PopenWrite { child, stdin } => { + stdin + .shutdown() + .await + .map_err(|error| VmError::HostError(format!("io close failed: {error}")))?; + wait_for_child(child, reason).await?; + } + } + Ok(()) +} + +async fn wait_for_child(child: &mut Child, reason: CancellationReason) -> VmResult<()> { + if !matches!(reason, CancellationReason::ResourceClosed) { + let _ = child.start_kill(); + } + match tokio::time::timeout(Duration::from_secs(1), child.wait()).await { + Ok(Ok(_)) => Ok(()), + Ok(Err(error)) => Err(VmError::HostError(format!( + "io process cleanup failed: {error}" + ))), + Err(_) => { + let _ = child.start_kill(); + child + .wait() + .await + .map(|_| ()) + .map_err(|error| VmError::HostError(format!("io process cleanup failed: {error}"))) + } + } +} + +fn start_close_io_handle(mut handle: IoHandle, _reason: CancellationReason) -> VmResult<()> { + match &mut handle { + IoHandle::File(_) => {} + IoHandle::PopenRead { child, .. } | IoHandle::PopenWrite { child, .. } => { + child.start_kill().map_err(|error| { + VmError::HostError(format!("io process cleanup failed: {error}")) + })?; + } + } + Ok(()) +} + +fn terminate_process_id(process_id: u32, reason: CancellationReason) -> VmResult<()> { + if process_id == 0 || matches!(reason, CancellationReason::ResourceClosed) { + return Ok(()); + } + #[cfg(unix)] + unsafe { + libc::kill(-(process_id as i32), libc::SIGKILL); + } + #[cfg(windows)] + { + let _ = process_id; + } + Ok(()) +} diff --git a/src/builtins/runtime/io.rs b/src/builtins/runtime/io/blocking.rs similarity index 91% rename from src/builtins/runtime/io.rs rename to src/builtins/runtime/io/blocking.rs index f7038c89..6c3bb0bf 100644 --- a/src/builtins/runtime/io.rs +++ b/src/builtins/runtime/io/blocking.rs @@ -15,13 +15,13 @@ use std::os::unix::process::CommandExt; use futures_channel::oneshot; use pd_host_function::pd_host_function; -use super::HostCallResult; -use super::cancellation::{CancellationReason, OperationId, OperationOwner}; -use super::error::{RuntimeError, RuntimeErrorCode}; -use super::resource::{ResourceHandle, ResourceTypeId}; +use super::super::HostCallResult; +use super::super::cancellation::{CancellationReason, OperationId, OperationOwner}; +use super::super::error::{RuntimeError, RuntimeErrorCode}; +use super::super::resource::{ResourceHandle, ResourceTypeId}; use crate::vm::{CallReturn, HostOpId, Value, Vm, VmError, VmResult}; -pub(super) enum IoHandle { +pub(crate) enum IoHandle { File(std::fs::File), PopenRead { child: Child }, PopenWrite { child: Child }, @@ -135,7 +135,7 @@ impl Drop for IoAsyncCompletion { } } -pub(super) fn poll_builtin_io_op( +pub(crate) fn poll_builtin_io_op( vm: &mut Vm, op_id: HostOpId, cx: &mut Context<'_>, @@ -168,10 +168,14 @@ pub(super) fn poll_builtin_io_op( match poll_result { Poll::Pending => Poll::Pending, Poll::Ready(Ok(mut completion)) => { - let _ = super::close_runtime_resource(vm, callback, CancellationReason::ResourceClosed); + let _ = super::super::close_runtime_resource( + vm, + callback, + CancellationReason::ResourceClosed, + ); if let Some(closed_handle) = completion.closed_handle - && let Err(error) = super::close_runtime_resource( + && let Err(error) = super::super::close_runtime_resource( vm, closed_handle, CancellationReason::ResourceClosed, @@ -190,7 +194,8 @@ pub(super) fn poll_builtin_io_op( )) } Poll::Ready(Err(_)) => { - let _ = super::close_runtime_resource(vm, callback, CancellationReason::Requested); + let _ = + super::super::close_runtime_resource(vm, callback, CancellationReason::Requested); Poll::Ready(Err(VmError::HostError(format!( "builtin io op {op_id} was cancelled", )))) @@ -200,7 +205,7 @@ pub(super) fn poll_builtin_io_op( /// Opens a file handle for runtime I/O. #[pd_host_function(name = "io::open")] -pub(super) fn builtin_io_open( +pub(crate) fn builtin_io_open( vm: &mut Vm, path: &str, mode: &str, @@ -260,7 +265,7 @@ pub(super) fn builtin_io_open( /// Starts a child process and returns a process-backed handle. #[pd_host_function(name = "io::popen")] -pub(super) fn builtin_io_popen( +pub(crate) fn builtin_io_popen( vm: &mut Vm, command: &str, mode: &str, @@ -270,12 +275,7 @@ pub(super) fn builtin_io_popen( "unsupported io_popen mode '{mode}', expected r or w" ))); } - if vm - .host - .io_policy - .as_ref() - .is_some_and(|policy| !policy.allow_process) - { + if super::io_policy(vm).is_some_and(|policy| !policy.allow_process) { return Err(VmError::HostError( "io_popen requires the process capability".to_string(), )); @@ -317,12 +317,8 @@ pub(super) fn builtin_io_popen( /// Reads all remaining text from an I/O handle. #[pd_host_function(name = "io::read_all")] -pub(super) fn builtin_io_read_all(vm: &mut Vm, handle_id: i64) -> VmResult> { - let max_read_bytes = vm - .host - .io_policy - .as_ref() - .map(|policy| policy.max_read_bytes); +pub(crate) fn builtin_io_read_all(vm: &mut Vm, handle_id: i64) -> VmResult> { + let max_read_bytes = super::io_policy(vm).map(|policy| policy.max_read_bytes); let handle = resource_handle(handle_id)?; let resource = io_resource_for_handle(vm, handle)?; let op_id = schedule_io_task(vm, Some(handle), move || { @@ -358,15 +354,11 @@ pub(super) fn builtin_io_read_all(vm: &mut Vm, handle_id: i64) -> VmResult VmResult> { - let max_read_bytes = vm - .host - .io_policy - .as_ref() - .map(|policy| policy.max_read_bytes); + let max_read_bytes = super::io_policy(vm).map(|policy| policy.max_read_bytes); let handle = resource_handle(handle_id)?; let resource = io_resource_for_handle(vm, handle)?; let op_id = schedule_io_task(vm, Some(handle), move || { @@ -394,12 +386,12 @@ pub(super) fn builtin_io_read_line( /// Writes text to an I/O handle. #[pd_host_function(name = "io::write")] -pub(super) fn builtin_io_write( +pub(crate) fn builtin_io_write( vm: &mut Vm, handle_id: i64, text: &str, ) -> VmResult> { - if let Some(policy) = vm.host.io_policy.as_ref() + if let Some(policy) = super::io_policy(vm) && text.len() > policy.max_write_bytes { return Err(VmError::HostError(format!( @@ -439,7 +431,7 @@ pub(super) fn builtin_io_write( /// Flushes buffered output for an I/O handle. #[pd_host_function(name = "io::flush")] -pub(super) fn builtin_io_flush(vm: &mut Vm, handle_id: i64) -> VmResult> { +pub(crate) fn builtin_io_flush(vm: &mut Vm, handle_id: i64) -> VmResult> { let handle = resource_handle(handle_id)?; let resource = io_resource_for_handle(vm, handle)?; let op_id = schedule_io_task(vm, Some(handle), move || { @@ -467,7 +459,7 @@ pub(super) fn builtin_io_flush(vm: &mut Vm, handle_id: i64) -> VmResult VmResult> { +pub(crate) fn builtin_io_close(vm: &mut Vm, handle_id: i64) -> VmResult> { let handle = resource_handle(handle_id)?; let resource = io_resource_for_handle(vm, handle)?; let op_id = schedule_io_task(vm, Some(handle), move || { @@ -486,7 +478,7 @@ pub(super) fn builtin_io_close(vm: &mut Vm, handle_id: i64) -> VmResult VmResult> { +pub(crate) fn builtin_io_exists(vm: &mut Vm, path: &str) -> VmResult> { let path = authorize_io_path(vm, path, false)?; let op_id = schedule_io_task(vm, None, move || { IoAsyncCompletion::result(Ok(CallReturn::one(Value::Bool(path.exists())))) @@ -496,7 +488,7 @@ pub(super) fn builtin_io_exists(vm: &mut Vm, path: &str) -> VmResult VmResult { let requested = PathBuf::from(path); - let Some(policy) = vm.host.io_policy.as_ref() else { + let Some(policy) = super::io_policy(vm) else { return Ok(requested); }; if writes && !policy.allow_write { @@ -641,49 +633,29 @@ fn schedule_io_task( }; operation.set_payload(callback); - if let Err(error) = std::thread::Builder::new() - .name("pd-vm-io".to_string()) - .spawn(move || { - let completion = if let Some(reason) = worker_token.reason() { - IoAsyncCompletion::result(Err(VmError::HostError(format!( - "io operation cancelled: {reason:?}" - )))) - } else { - task() - }; - match &completion.result { - Ok(_) => { - let _ = worker_operation.complete(); - } - Err(error) => { - let _ = worker_operation.fail( - RuntimeError::new( - RuntimeErrorCode::OperationFailed, - "io::operation", - error.to_string(), - ) - .with_value(op_id), - ); - } - } - let _ = sender.send(completion); - }) - { - let runtime_error = RuntimeError::new( - RuntimeErrorCode::OperationFailed, - "io::schedule", - format!("failed to spawn io task: {error}"), - ) - .with_value(op_id); - let _ = super::close_runtime_resource(vm, callback, CancellationReason::Requested); - let _ = vm - .host - .runtime_operations - .fail(operation.id(), runtime_error); - return Err(VmError::HostError(format!( - "failed to spawn io task: {error}" - ))); + let completion = if let Some(reason) = worker_token.reason() { + IoAsyncCompletion::result(Err(VmError::HostError(format!( + "io operation cancelled: {reason:?}" + )))) + } else { + task() + }; + match &completion.result { + Ok(_) => { + let _ = worker_operation.complete(); + } + Err(error) => { + let _ = worker_operation.fail( + RuntimeError::new( + RuntimeErrorCode::OperationFailed, + "io::operation", + error.to_string(), + ) + .with_value(op_id), + ); + } } + let _ = sender.send(completion); Ok(op_id) } @@ -839,7 +811,7 @@ mod windows_process_tree { fn CloseHandle(handle: Handle) -> i32; } - pub(super) fn terminate(root_process_id: u32) -> VmResult<()> { + pub(crate) fn terminate(root_process_id: u32) -> VmResult<()> { let descendants = match descendant_processes(root_process_id) { Ok(descendants) => descendants, Err(snapshot_error) => { diff --git a/src/builtins/runtime/io/mod.rs b/src/builtins/runtime/io/mod.rs new file mode 100644 index 00000000..4dd722c1 --- /dev/null +++ b/src/builtins/runtime/io/mod.rs @@ -0,0 +1,66 @@ +use super::borrow_arg; +#[cfg(feature = "async")] +use super::{CallOutcome, CaptureAsyncHostContext, return_one}; +use crate::vm::Vm; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct IoPolicy { + pub allowed_roots: Vec, + pub allow_write: bool, + pub allow_process: bool, + pub max_read_bytes: usize, + pub max_write_bytes: usize, +} + +impl Default for IoPolicy { + fn default() -> Self { + Self { + allowed_roots: Vec::new(), + allow_write: false, + allow_process: false, + max_read_bytes: 1024 * 1024, + max_write_bytes: 1024 * 1024, + } + } +} + +struct IoHostState { + policy: IoPolicy, +} + +/// I/O host configuration owned by the I/O host implementation. +pub trait IoHostExt { + fn configure_io(&mut self, policy: IoPolicy); + fn clear_io_configuration(&mut self); +} + +impl IoHostExt for Vm { + fn configure_io(&mut self, mut policy: IoPolicy) { + policy.allowed_roots.sort(); + policy.allowed_roots.dedup(); + self.host.set_host_function_state(IoHostState { policy }); + } + + fn clear_io_configuration(&mut self) { + self.host.remove_host_function_state::(); + } +} + +pub(super) fn io_policy(vm: &Vm) -> Option { + vm.host + .host_function_state::() + .map(|state| state.policy.clone()) + .or_else(|| (!vm.host.default_builtin_capabilities_enabled()).then(IoPolicy::default)) +} + +#[cfg(all(feature = "async", not(target_arch = "wasm32")))] +mod async_io; +#[cfg(all(not(feature = "async"), not(target_arch = "wasm32")))] +mod blocking; + +#[cfg(target_arch = "wasm32")] +pub(super) use super::io_wasm::*; +#[cfg(all(feature = "async", not(target_arch = "wasm32")))] +pub(super) use async_io::*; +#[cfg(all(not(feature = "async"), not(target_arch = "wasm32")))] +pub(super) use blocking::*; diff --git a/src/builtins/runtime/mod.rs b/src/builtins/runtime/mod.rs index 40ead3ef..96da098e 100644 --- a/src/builtins/runtime/mod.rs +++ b/src/builtins/runtime/mod.rs @@ -5,7 +5,7 @@ use std::task::{Context, Poll}; use crate::builtins::BuiltinFunction; use crate::vm::{CallOutcome, CallReturn, HostOpId, Value, Vm, VmResult}; #[cfg(feature = "async")] -use crate::vm::{CaptureAsyncHostContext, VmError}; +use crate::vm::{CaptureAsyncHostContext, HostFutureOutput, VmError}; use self::cancellation::{CancellationReason, OperationId, OperationOwner, OperationState}; use self::error::{RuntimeError, RuntimeErrorCode}; @@ -16,6 +16,7 @@ use self::resource::ResourceTypeId; type RuntimeOperationPoller = fn(&mut Vm, HostOpId, &mut Context<'_>) -> Poll>; const RUNTIME_OPERATION_POLLERS: &[(OperationOwner, RuntimeOperationPoller)] = &[ + #[cfg(not(feature = "async"))] (OperationOwner::Io, io::poll_builtin_io_op), #[cfg(feature = "sqlite")] (OperationOwner::Sqlite, sqlite::poll_pending_op), @@ -31,7 +32,6 @@ pub(crate) mod error; pub(crate) mod event; mod host; mod http; -#[cfg(not(target_arch = "wasm32"))] mod io; #[cfg(target_arch = "wasm32")] mod io_wasm; @@ -46,11 +46,10 @@ pub(crate) mod resource; mod sqlite; mod typed; -#[cfg(target_arch = "wasm32")] -use io_wasm as io; - -pub use http::HttpConfig; -pub(crate) use http::HttpState; +pub use http::{HttpConfig, HttpHostExt}; +pub use io::{IoHostExt, IoPolicy}; +#[cfg(feature = "sqlite")] +pub use sqlite::{SqliteHostExt, SqliteLimits, SqlitePolicy}; pub use typed::HostCallResult; use typed::{ AnyValue, IntoBuiltinCallOutcome, IntoHostCallOutcome, NumberValue, UnknownValue, VmArray, diff --git a/src/builtins/runtime/resource.rs b/src/builtins/runtime/resource.rs index f9d9ed0e..49500e07 100644 --- a/src/builtins/runtime/resource.rs +++ b/src/builtins/runtime/resource.rs @@ -36,6 +36,7 @@ impl ResourceTypeId { #[cfg_attr(not(feature = "sqlite"), allow(dead_code))] pub const SQLITE_CONNECTION: Self = Self(5); + #[cfg_attr(feature = "async", allow(dead_code))] pub const CALLBACK: Self = Self(6); pub const fn raw(self) -> u16 { @@ -184,6 +185,7 @@ impl ResourceArena { }) } + #[cfg_attr(feature = "async", allow(dead_code))] pub fn insert( &mut self, resource_type: ResourceTypeId, @@ -255,6 +257,7 @@ impl ResourceArena { .ok_or_else(|| type_mismatch(handle, expected_type)) } + #[cfg_attr(feature = "async", allow(dead_code))] pub fn get_mut( &mut self, handle: ResourceHandle, @@ -429,6 +432,7 @@ impl ResourceArena { Ok(slot) } + #[cfg_attr(feature = "async", allow(dead_code))] fn active_slot_mut( &mut self, handle: ResourceHandle, diff --git a/src/builtins/runtime/sqlite.rs b/src/builtins/runtime/sqlite.rs index 05b1a87a..8d6ae781 100644 --- a/src/builtins/runtime/sqlite.rs +++ b/src/builtins/runtime/sqlite.rs @@ -18,11 +18,99 @@ use super::error::{RuntimeError, RuntimeErrorCode}; use super::resource::{ResourceHandle, ResourceTypeId}; use super::typed::{VmArrayRef, VmMapRef}; use super::{HostCallResult, VmMap}; -use crate::vm::{CallReturn, HostOpId, SqliteLimits, Value, Vm, VmError, VmResult}; +use crate::vm::{CallReturn, HostOpId, Value, Vm, VmError, VmResult}; const SQLITE_PROGRESS_STEPS: i32 = 1_000; const SQLITE_CLOSE_GRACE: Duration = Duration::from_millis(100); +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SqliteLimits { + pub max_connections: usize, + pub max_statements: usize, + pub max_rows: usize, + pub max_columns: usize, + pub max_result_bytes: usize, + pub max_statement_bytes: usize, + pub max_parameters: usize, + pub max_parameter_bytes: usize, + pub max_pending_operations: usize, + pub max_transaction_ms: u64, + pub busy_timeout_ms: u64, +} + +impl Default for SqliteLimits { + fn default() -> Self { + Self { + max_connections: 16, + max_statements: 128, + max_rows: 1_000, + max_columns: 128, + max_result_bytes: 4 * 1024 * 1024, + max_statement_bytes: 1024 * 1024, + max_parameters: 128, + max_parameter_bytes: 1024 * 1024, + max_pending_operations: 32, + max_transaction_ms: 5_000, + busy_timeout_ms: 5_000, + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SqlitePolicy { + pub database_root: Option, + pub allow_unsafe_sql: bool, + pub limits: SqliteLimits, +} + +struct SqliteHostState { + policy: SqlitePolicy, +} + +/// SQLite host configuration owned by the SQLite host implementation. +#[allow(dead_code)] +pub trait SqliteHostExt { + fn configure_sqlite(&mut self, policy: SqlitePolicy); + fn clear_sqlite_configuration(&mut self); +} + +impl SqliteHostExt for Vm { + fn configure_sqlite(&mut self, policy: SqlitePolicy) { + super::cancel_operations_by_owner( + self, + OperationOwner::Sqlite, + CancellationReason::ResourceClosed, + ); + super::close_resources_by_type( + self, + ResourceTypeId::SQLITE_CONNECTION, + CancellationReason::ResourceClosed, + ); + self.host + .set_host_function_state(SqliteHostState { policy }); + } + + fn clear_sqlite_configuration(&mut self) { + super::cancel_operations_by_owner( + self, + OperationOwner::Sqlite, + CancellationReason::ResourceClosed, + ); + super::close_resources_by_type( + self, + ResourceTypeId::SQLITE_CONNECTION, + CancellationReason::ResourceClosed, + ); + self.host.remove_host_function_state::(); + } +} + +fn sqlite_policy(vm: &Vm) -> SqlitePolicy { + vm.host + .host_function_state::() + .map_or_else(SqlitePolicy::default, |state| state.policy.clone()) +} + /// Returns the affected-row count from a SQLite result envelope. #[pd_host_function(name = "sqlite::rows_affected")] pub(super) fn builtin_sqlite_rows_affected_impl(value: VmMapRef<'_>) -> VmResult { @@ -241,12 +329,8 @@ fn parse_open_options(vm: &Vm, options: &VmMap) -> VmResult { ))); } }; - let configured_root = vm - .host - .sqlite_policy - .database_root - .as_deref() - .map(PathBuf::from); + let policy = sqlite_policy(vm); + let configured_root = policy.database_root.as_deref().map(PathBuf::from); if let Some(requested_root) = optional_string(options, "root")? { let requested_root = PathBuf::from(requested_root); if configured_root.as_ref() != Some(&requested_root) { @@ -260,13 +344,13 @@ fn parse_open_options(vm: &Vm, options: &VmMap) -> VmResult { "SQLite database root is not configured".to_string(), )); } - let limits = parse_limits(map_value(options, "limits"), vm.host.sqlite_policy.limits)?; + let limits = parse_limits(map_value(options, "limits"), policy.limits)?; Ok(OpenOptions { path, mode, root: configured_root, limits, - allow_unsafe_sql: vm.host.sqlite_policy.allow_unsafe_sql, + allow_unsafe_sql: policy.allow_unsafe_sql, }) } diff --git a/src/builtins/runtime/typed.rs b/src/builtins/runtime/typed.rs index 44f57684..9162807a 100644 --- a/src/builtins/runtime/typed.rs +++ b/src/builtins/runtime/typed.rs @@ -130,6 +130,15 @@ impl<'a> FromVmValue<'a> for &'a str { } } +impl FromVmValue<'_> for String { + fn from_vm_value(value: &Value, _label: &str) -> VmResult { + match value { + Value::String(text) => Ok(text.to_string()), + _ => Err(VmError::TypeMismatch("string")), + } + } +} + impl<'a> FromVmValue<'a> for &'a [u8] { fn from_vm_value(value: &'a Value, _label: &str) -> VmResult { match value { @@ -462,6 +471,17 @@ where } } +impl IntoBuiltinCallOutcome for CallOutcome { + fn into_builtin_call_outcome(self) -> BuiltinCallOutcome { + match self { + CallOutcome::Return(values) => BuiltinCallOutcome::Return(values), + CallOutcome::Halt => BuiltinCallOutcome::Halt, + CallOutcome::Pending(op_id) => BuiltinCallOutcome::Pending(op_id), + CallOutcome::Yield => unreachable!("async builtin wrappers cannot return Yield"), + } + } +} + impl IntoBuiltinCallOutcome for HostCallResult where T: IntoVmValue, diff --git a/src/lib.rs b/src/lib.rs index bdabbe6c..7fa63e1d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,9 +24,11 @@ pub use assembler::{AsmParseError, Assembler, AssemblerError, BytecodeBuilder, a #[cfg(feature = "runtime")] pub use builtins::runtime::HostCallResult; #[cfg(feature = "runtime")] -pub use builtins::runtime::HttpConfig; -#[cfg(feature = "runtime")] pub use builtins::runtime::print::{PrintHostFunction, PrintlnHostFunction, format_value}; +#[cfg(feature = "runtime")] +pub use builtins::runtime::{HttpConfig, HttpHostExt, IoHostExt, IoPolicy}; +#[cfg(feature = "sqlite")] +pub use builtins::runtime::{SqliteHostExt, SqliteLimits, SqlitePolicy}; pub use builtins::{ BUILTIN_CATALOG, BuiltinFunction, BuiltinNamespaceMemberSpec, BuiltinNamespaceSpec, CallableDef, CallableParam, CallableParamType, CallableSignature, HostExecution, @@ -91,13 +93,12 @@ pub use vm::{ AotArtifactError, CallOutcome, CallReturn, CancellationReason, CapabilityProfile, CapabilityProfileBuilder, DEFAULT_MAX_SCRIPT_CALL_DEPTH, EpochCheckpoint, EpochHandle, FuelCheckpoint, HostArgsFunction, HostAsyncBridge, HostBindingPlan, HostFunction, - HostFunctionRegistry, HostOpId, HostStackFunction, IntoScriptValue, IoPolicy, - QueuedScriptInvocation, ScriptArgs, ScriptCallback, ScriptResult, StaticHostArgsFunction, - StaticHostFunction, StaticHostStackFunction, Store, Vm, VmError, VmResult, VmStatus, - VmYieldReason, + HostFunctionRegistry, HostFuture, HostFutureOutput, HostOpId, HostStackFunction, + IntoScriptValue, QueuedScriptInvocation, ScriptArgs, ScriptCallback, ScriptResult, + StaticHostArgsFunction, StaticHostFunction, StaticHostStackFunction, Store, Vm, VmError, + VmResult, VmStatus, VmYieldReason, }; -#[cfg(feature = "sqlite")] -pub use vm::{SqliteLimits, SqlitePolicy}; + #[cfg(feature = "runtime")] pub use vmbc::{ DisassembleOptions, ValidationError, WireError, decode_program, disassemble_program, diff --git a/src/vm/async_host/mod.rs b/src/vm/async_host/mod.rs index b81c5085..68ec8e80 100644 --- a/src/vm/async_host/mod.rs +++ b/src/vm/async_host/mod.rs @@ -5,10 +5,61 @@ use std::task::{Context, Poll, Wake, Waker}; use super::*; -pub type HostFuture = Pin> + Send + 'static>>; +type HostVmCompletion = Box VmResult + Send + 'static>; + +pub enum HostFutureOutput { + Return(T), + VmCompletion(HostVmCompletion), +} + +impl HostFutureOutput { + pub fn returning(value: T) -> Self { + Self::Return(value) + } + + pub fn complete(completion: impl FnOnce(&mut Vm) -> VmResult + Send + 'static) -> Self { + Self::VmCompletion(Box::new(completion)) + } + + pub fn map( + self, + map: impl FnOnce(T) -> U + Send + 'static, + ) -> HostFutureOutput + where + T: Send + 'static, + { + match self { + Self::Return(value) => HostFutureOutput::Return(map(value)), + Self::VmCompletion(completion) => { + HostFutureOutput::VmCompletion(Box::new(move |vm| completion(vm).map(map))) + } + } + } +} + +impl HostFutureOutput { + fn finish(self, vm: &mut Vm) -> VmResult { + match self { + Self::Return(values) => Ok(values), + Self::VmCompletion(completion) => completion(vm), + } + } +} + +impl From for HostFutureOutput { + fn from(values: CallReturn) -> Self { + Self::Return(values) + } +} + +pub type HostFuture = Pin> + Send + 'static>>; pub trait CaptureAsyncHostContext: Send + 'static + Sized { fn capture(vm: &mut Vm) -> VmResult; + + fn capture_with_args(vm: &mut Vm, _args: &[Value]) -> VmResult { + Self::capture(vm) + } } pub trait HostAsyncBridge: Send { @@ -20,6 +71,15 @@ pub trait HostAsyncBridge: Send { fn poll_op(&mut self, op_id: HostOpId, cx: &mut Context<'_>) -> Poll>; + fn poll_submitted_op( + &mut self, + op_id: HostOpId, + cx: &mut Context<'_>, + ) -> Poll> { + self.poll_op(op_id, cx) + .map(|result| result.map(HostFutureOutput::Return)) + } + fn cancel_op(&mut self, _op_id: HostOpId) {} fn cancel_op_with_reason(&mut self, op_id: HostOpId, _reason: CancellationReason) { @@ -67,6 +127,7 @@ impl Vm { VmError::HostError("async host function requires a host async bridge".to_string()) })?; bridge.submit_op(op_id, future)?; + self.host.submitted_host_ops.insert(op_id); Ok(CallOutcome::Pending(op_id)) } @@ -102,6 +163,7 @@ impl Vm { if let Some(bridge) = self.host.async_bridge.as_mut() { bridge.cancel_op_with_reason(waiting.op_id, reason); } + self.host.submitted_host_ops.remove(&waiting.op_id); let _ = self.host.runtime_operations.cancel(operation_id, reason); } else { crate::builtins::runtime::cancel_builtin_io_op_with_reason(self, waiting.op_id, reason); @@ -140,6 +202,11 @@ impl Vm { .runtime_operations .complete(operation_id) .map_err(|error| VmError::HostError(error.to_string()))?; + if self.host.submitted_host_ops.remove(&op_id) + && let Some(bridge) = self.host.async_bridge.as_mut() + { + bridge.cancel_op(op_id); + } self.complete_waiting_host_op(op_id, values.into()) } @@ -169,25 +236,53 @@ impl Vm { )))); } }; - unsafe { (&mut *bridge_ptr).poll_op(waiting.op_id, cx) } + if self.host.submitted_host_ops.contains(&waiting.op_id) { + unsafe { (&mut *bridge_ptr).poll_submitted_op(waiting.op_id, cx) } + } else { + unsafe { (&mut *bridge_ptr).poll_op(waiting.op_id, cx) } + .map(|result| result.map(HostFutureOutput::Return)) + } } else { crate::builtins::runtime::poll_builtin_io_op(self, waiting.op_id, cx) + .map(|result| result.map(HostFutureOutput::Return)) }; match poll_result { Poll::Pending => Poll::Pending, - Poll::Ready(Ok(values)) => { + Poll::Ready(Ok(output)) => { + let values = match output.finish(self) { + Ok(values) => values, + Err(err) => { + if host_bridge_owned { + self.host.submitted_host_ops.remove(&waiting.op_id); + let runtime_error = crate::builtins::runtime::error::RuntimeError::new( + crate::builtins::runtime::error::RuntimeErrorCode::OperationFailed, + "runtime::host_bridge", + err.to_string(), + ) + .with_value(waiting.op_id); + let _ = self + .host + .runtime_operations + .fail(operation_id, runtime_error); + } + self.instance.waiting_host_op = None; + return Poll::Ready(Err(err)); + } + }; if host_bridge_owned { self.host .runtime_operations .complete(operation_id) .map_err(|error| VmError::HostError(error.to_string()))?; + self.host.submitted_host_ops.remove(&waiting.op_id); } self.complete_waiting_host_op(waiting.op_id, values)?; Poll::Ready(Ok(())) } Poll::Ready(Err(err)) => { if host_bridge_owned { + self.host.submitted_host_ops.remove(&waiting.op_id); let runtime_error = crate::builtins::runtime::error::RuntimeError::new( crate::builtins::runtime::error::RuntimeErrorCode::OperationFailed, "runtime::host_bridge", diff --git a/src/vm/capability.rs b/src/vm/capability.rs index 2618d2fb..9c60be11 100644 --- a/src/vm/capability.rs +++ b/src/vm/capability.rs @@ -1,32 +1,8 @@ -#[cfg(feature = "sqlite")] -use super::SqlitePolicy; use crate::builtins::BuiltinFunction; -use crate::builtins::runtime::HttpConfig; const FNV_OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325; const FNV_PRIME: u64 = 0x0000_0100_0000_01b3; -const PROFILE_VERSION: &[u8] = b"rustscript-capability-profile-v1"; - -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct IoPolicy { - pub allowed_roots: Vec, - pub allow_write: bool, - pub allow_process: bool, - pub max_read_bytes: usize, - pub max_write_bytes: usize, -} - -impl Default for IoPolicy { - fn default() -> Self { - Self { - allowed_roots: Vec::new(), - allow_write: false, - allow_process: false, - max_read_bytes: 1024 * 1024, - max_write_bytes: 1024 * 1024, - } - } -} +const PROFILE_VERSION: &[u8] = b"rustscript-capability-profile-v2"; /// Immutable authorization policy for privileged builtin calls and host imports. #[derive(Clone, Debug, PartialEq, Eq)] @@ -35,10 +11,6 @@ pub struct CapabilityProfile { allow_all_host_imports: bool, allowed_builtin_calls: Vec, allowed_host_imports: Vec, - http_policy: Option, - io_policy: Option, - #[cfg(feature = "sqlite")] - sqlite_policy: Option, fingerprint: u64, } @@ -80,19 +52,6 @@ impl CapabilityProfile { .is_ok() } - pub fn http_policy(&self) -> Option<&HttpConfig> { - self.http_policy.as_ref() - } - - pub fn io_policy(&self) -> Option<&IoPolicy> { - self.io_policy.as_ref() - } - - #[cfg(feature = "sqlite")] - pub fn sqlite_policy(&self) -> Option<&SqlitePolicy> { - self.sqlite_policy.as_ref() - } - pub(crate) fn allowed_builtin_calls(&self) -> &[u16] { &self.allowed_builtin_calls } @@ -111,10 +70,6 @@ impl CapabilityProfile { allow_all_host_imports: self.allow_all_host_imports, allowed_builtin_calls: self.allowed_builtin_calls.clone(), allowed_host_imports: self.allowed_host_imports.clone(), - http_policy: self.http_policy.clone(), - io_policy: self.io_policy.clone(), - #[cfg(feature = "sqlite")] - sqlite_policy: self.sqlite_policy.clone(), }; builder.allowed_builtin_calls.push(builtin.call_index()); builder.build() @@ -126,10 +81,6 @@ impl CapabilityProfile { allow_all_host_imports: self.allow_all_host_imports, allowed_builtin_calls: self.allowed_builtin_calls.clone(), allowed_host_imports: self.allowed_host_imports.clone(), - http_policy: self.http_policy.clone(), - io_policy: self.io_policy.clone(), - #[cfg(feature = "sqlite")] - sqlite_policy: self.sqlite_policy.clone(), }; builder.allowed_host_imports.push(name.to_string()); builder.build() @@ -148,10 +99,6 @@ pub struct CapabilityProfileBuilder { allow_all_host_imports: bool, allowed_builtin_calls: Vec, allowed_host_imports: Vec, - http_policy: Option, - io_policy: Option, - #[cfg(feature = "sqlite")] - sqlite_policy: Option, } impl CapabilityProfileBuilder { @@ -165,58 +112,22 @@ impl CapabilityProfileBuilder { self } - pub fn http_policy(mut self, policy: HttpConfig) -> Self { - self.http_policy = Some(policy); - self - } - - pub fn io_policy(mut self, policy: IoPolicy) -> Self { - self.io_policy = Some(policy); - self - } - - #[cfg(feature = "sqlite")] - pub fn sqlite_policy(mut self, policy: SqlitePolicy) -> Self { - self.sqlite_policy = Some(policy); - self - } - pub fn build(mut self) -> CapabilityProfile { self.allowed_builtin_calls.sort_unstable(); self.allowed_builtin_calls.dedup(); self.allowed_host_imports.sort(); self.allowed_host_imports.dedup(); - if let Some(policy) = self.http_policy.as_mut() { - policy.allowed_schemes.sort(); - policy.allowed_schemes.dedup(); - policy.allowed_hosts.sort(); - policy.allowed_hosts.dedup(); - policy.allowed_ports.sort_unstable(); - policy.allowed_ports.dedup(); - } - if let Some(policy) = self.io_policy.as_mut() { - policy.allowed_roots.sort(); - policy.allowed_roots.dedup(); - } let fingerprint = fingerprint( self.allow_all_builtins, self.allow_all_host_imports, &self.allowed_builtin_calls, &self.allowed_host_imports, - self.http_policy.as_ref(), - self.io_policy.as_ref(), - #[cfg(feature = "sqlite")] - self.sqlite_policy.as_ref(), ); CapabilityProfile { allow_all_builtins: self.allow_all_builtins, allow_all_host_imports: self.allow_all_host_imports, allowed_builtin_calls: self.allowed_builtin_calls, allowed_host_imports: self.allowed_host_imports, - http_policy: self.http_policy, - io_policy: self.io_policy, - #[cfg(feature = "sqlite")] - sqlite_policy: self.sqlite_policy, fingerprint, } } @@ -227,9 +138,6 @@ fn fingerprint( allow_all_host_imports: bool, builtin_calls: &[u16], host_imports: &[String], - http_policy: Option<&HttpConfig>, - io_policy: Option<&IoPolicy>, - #[cfg(feature = "sqlite")] sqlite_policy: Option<&SqlitePolicy>, ) -> u64 { let mut value = FNV_OFFSET_BASIS; update_fingerprint(&mut value, PROFILE_VERSION); @@ -249,91 +157,12 @@ fn fingerprint( update_fingerprint(&mut value, &(name.len() as u64).to_le_bytes()); update_fingerprint(&mut value, name.as_bytes()); } - match http_policy { - None => update_fingerprint(&mut value, &[0]), - Some(policy) => { - update_fingerprint(&mut value, &[1]); - update_string_list(&mut value, &policy.allowed_schemes); - update_string_list(&mut value, &policy.allowed_hosts); - update_fingerprint( - &mut value, - &(policy.allowed_ports.len() as u64).to_le_bytes(), - ); - for port in &policy.allowed_ports { - update_fingerprint(&mut value, &port.to_le_bytes()); - } - for limit in [ - policy.max_redirects as u64, - policy.max_request_body_bytes as u64, - policy.max_response_body_bytes as u64, - policy.connect_timeout.as_secs(), - u64::from(policy.connect_timeout.subsec_nanos()), - policy.request_timeout.as_secs(), - u64::from(policy.request_timeout.subsec_nanos()), - ] { - update_fingerprint(&mut value, &limit.to_le_bytes()); - } - update_fingerprint(&mut value, &[u8::from(policy.allow_private_ips)]); - } - } - match io_policy { - None => update_fingerprint(&mut value, &[0]), - Some(policy) => { - update_fingerprint(&mut value, &[1]); - update_string_list(&mut value, &policy.allowed_roots); - update_fingerprint( - &mut value, - &[u8::from(policy.allow_write), u8::from(policy.allow_process)], - ); - update_fingerprint(&mut value, &(policy.max_read_bytes as u64).to_le_bytes()); - update_fingerprint(&mut value, &(policy.max_write_bytes as u64).to_le_bytes()); - } - } - #[cfg(feature = "sqlite")] - match sqlite_policy { - None => update_fingerprint(&mut value, &[0]), - Some(policy) => { - update_fingerprint(&mut value, &[1]); - match &policy.database_root { - None => update_fingerprint(&mut value, &[0]), - Some(root) => { - update_fingerprint(&mut value, &[1]); - update_fingerprint(&mut value, &(root.len() as u64).to_le_bytes()); - update_fingerprint(&mut value, root.as_bytes()); - } - } - update_fingerprint(&mut value, &[u8::from(policy.allow_unsafe_sql)]); - for limit in [ - policy.limits.max_connections as u64, - policy.limits.max_statements as u64, - policy.limits.max_rows as u64, - policy.limits.max_columns as u64, - policy.limits.max_result_bytes as u64, - policy.limits.max_statement_bytes as u64, - policy.limits.max_parameters as u64, - policy.limits.max_parameter_bytes as u64, - policy.limits.max_pending_operations as u64, - policy.limits.max_transaction_ms, - policy.limits.busy_timeout_ms, - ] { - update_fingerprint(&mut value, &limit.to_le_bytes()); - } - } - } value } -fn update_string_list(fingerprint: &mut u64, values: &[String]) { - update_fingerprint(fingerprint, &(values.len() as u64).to_le_bytes()); - for value in values { - update_fingerprint(fingerprint, &(value.len() as u64).to_le_bytes()); - update_fingerprint(fingerprint, value.as_bytes()); - } -} - -fn update_fingerprint(fingerprint: &mut u64, bytes: &[u8]) { +fn update_fingerprint(state: &mut u64, bytes: &[u8]) { for byte in bytes { - *fingerprint ^= u64::from(*byte); - *fingerprint = fingerprint.wrapping_mul(FNV_PRIME); + *state ^= u64::from(*byte); + *state = state.wrapping_mul(FNV_PRIME); } } diff --git a/src/vm/host.rs b/src/vm/host.rs index 2a5e55e6..676997b4 100644 --- a/src/vm/host.rs +++ b/src/vm/host.rs @@ -643,26 +643,6 @@ impl HostFunctionRegistry { )); } - if !plan.allow_default_host_capabilities { - match plan.capability_profile.http_policy() { - Some(policy) => vm.host.http_state.configure(policy.clone()), - None => vm.host.http_state.clear_configuration(), - } - vm.host.io_policy = Some( - plan.capability_profile - .io_policy() - .cloned() - .unwrap_or_default(), - ); - #[cfg(feature = "sqlite")] - { - vm.host.sqlite_policy = plan - .capability_profile - .sqlite_policy() - .cloned() - .unwrap_or_default(); - } - } vm.host.host_functions.reserve(plan.registry_slots.len()); for ®istry_slot in &plan.registry_slots { let entry = self @@ -1166,61 +1146,6 @@ impl Vm { .map_err(|error| VmError::HostError(error.to_string())) } - #[cfg(feature = "sqlite")] - pub fn configure_sqlite(&mut self, policy: crate::vm::SqlitePolicy) { - crate::builtins::runtime::cancel_operations_by_owner( - self, - crate::builtins::runtime::cancellation::OperationOwner::Sqlite, - crate::builtins::runtime::cancellation::CancellationReason::ResourceClosed, - ); - crate::builtins::runtime::close_resources_by_type( - self, - crate::builtins::runtime::resource::ResourceTypeId::SQLITE_CONNECTION, - crate::builtins::runtime::cancellation::CancellationReason::ResourceClosed, - ); - self.host.sqlite_policy = policy; - } - - #[cfg(feature = "sqlite")] - pub fn clear_sqlite_configuration(&mut self) { - crate::builtins::runtime::cancel_operations_by_owner( - self, - crate::builtins::runtime::cancellation::OperationOwner::Sqlite, - crate::builtins::runtime::cancellation::CancellationReason::ResourceClosed, - ); - crate::builtins::runtime::close_resources_by_type( - self, - crate::builtins::runtime::resource::ResourceTypeId::SQLITE_CONNECTION, - crate::builtins::runtime::cancellation::CancellationReason::ResourceClosed, - ); - self.host.sqlite_policy = crate::vm::SqlitePolicy::default(); - } - - pub fn configure_http(&mut self, config: crate::builtins::runtime::HttpConfig) { - self.host.http_state.configure(config); - } - - pub fn set_http_max_in_flight(&mut self, max_in_flight: usize) { - self.host.http_state.max_in_flight = max_in_flight; - } - - pub fn http_max_in_flight(&self) -> usize { - self.host.http_state.max_in_flight - } - - pub fn clear_http_configuration(&mut self) { - crate::builtins::runtime::cancel_operations_by_owner( - self, - crate::builtins::runtime::cancellation::OperationOwner::Http, - crate::builtins::runtime::cancellation::CancellationReason::Requested, - ); - self.host.http_state.clear_configuration(); - } - - pub fn http_is_configured(&self) -> bool { - self.host.http_state.is_configured() - } - /// Enables or disables implicit binding of built-in host functions. /// /// Disabling this makes the VM use only explicitly registered host functions. The default @@ -1390,7 +1315,17 @@ impl Vm { crate::builtins::runtime::BuiltinCallOutcome::Pending(op_id) => { self.instance.stack.truncate(arg_start); let resume_ip = self.call_resume_ip(call_ip)?; - self.set_waiting_registered_op(op_id)?; + if self.host.submitted_host_ops.contains(&op_id) { + if let Err(error) = self.set_waiting_host_op(op_id) { + self.host.submitted_host_ops.remove(&op_id); + if let Some(bridge) = self.host.async_bridge.as_mut() { + bridge.cancel_op(op_id); + } + return Err(error); + } + } else { + self.set_waiting_registered_op(op_id)?; + } self.instance.ip = resume_ip; Ok(HostCallExecOutcome::Pending(op_id)) } diff --git a/src/vm/host_runtime.rs b/src/vm/host_runtime.rs index 34f2f761..49b583ae 100644 --- a/src/vm/host_runtime.rs +++ b/src/vm/host_runtime.rs @@ -3,25 +3,23 @@ //! [`HostRuntime`] owns the host-facing capability surface: bound host //! functions and their symbol table, capability allow-lists, builtin //! overrides, resolved call slots, the opaque resource arena, the pending -//! operation registry, and the IO/HTTP/SQLite subsystem state plus the async -//! bridge and print sink. Interpreter state and run budgets live outside this -//! struct (see [`Instance`](super::instance::Instance) and +//! operation registry, a type-erased host-function state store, the async +//! bridge, and the print sink. Concrete IO/HTTP/SQLite state is defined and +//! interpreted only by those host modules. Interpreter state and run budgets +//! live outside this struct (see [`Instance`](super::instance::Instance) and //! [`RunContext`](super::run_context::RunContext)). //! -//! The unified host-lifecycle plan migrates individual subsystems behind this -//! shell; for now it groups their ownership and their reset/drop behavior. +//! The VM provides lifecycle storage without depending on host-specific state +//! types or configuration APIs. +use std::any::{Any, TypeId}; use std::collections::{HashMap, HashSet}; -use crate::builtins::runtime::HttpState; use crate::builtins::runtime::cancellation::{ CancellationReason, DEFAULT_MAX_PENDING_OPERATIONS, OperationRegistry, }; use crate::builtins::runtime::resource::{DEFAULT_MAX_RESOURCES, ResourceArena}; -use crate::vm::IoPolicy; -#[cfg(feature = "sqlite")] -use crate::vm::SqlitePolicy; use crate::vm::async_host::HostAsyncBridge; use crate::vm::host::VmHostFunction; @@ -48,11 +46,9 @@ pub(crate) struct HostRuntime { pub(crate) resolved_calls_dirty: bool, pub(crate) runtime_resources: ResourceArena, pub(crate) runtime_operations: OperationRegistry, - pub(crate) io_policy: Option, - #[cfg(feature = "sqlite")] - pub(crate) sqlite_policy: SqlitePolicy, - pub(crate) http_state: HttpState, + host_function_states: HashMap>, pub(crate) async_bridge: Option>, + pub(crate) submitted_host_ops: HashSet, pub(crate) runtime_print_sink: Option>, } @@ -76,11 +72,9 @@ impl HostRuntime { .expect("default runtime resource limit should be valid"), runtime_operations: OperationRegistry::with_limit(DEFAULT_MAX_PENDING_OPERATIONS) .expect("default runtime operation limit should be valid"), - io_policy: None, - #[cfg(feature = "sqlite")] - sqlite_policy: SqlitePolicy::default(), - http_state: HttpState::default(), + host_function_states: HashMap::new(), async_bridge: None, + submitted_host_ops: HashSet::new(), runtime_print_sink: None, } } @@ -96,7 +90,48 @@ impl HostRuntime { let _ = self .runtime_resources .close_all(CancellationReason::VmReset); - self.http_state.reset_for_reuse(); + self.submitted_host_ops.clear(); + } + + pub(crate) fn set_host_function_state(&mut self, state: T) + where + T: Any + Send, + { + self.host_function_states + .insert(TypeId::of::(), Box::new(state)); + } + + pub(crate) fn host_function_state(&self) -> Option<&T> + where + T: Any + Send, + { + self.host_function_states + .get(&TypeId::of::())? + .downcast_ref() + } + + pub(crate) fn host_function_state_mut(&mut self) -> Option<&mut T> + where + T: Any + Send, + { + self.host_function_states + .get_mut(&TypeId::of::())? + .downcast_mut() + } + + pub(crate) fn remove_host_function_state(&mut self) -> Option + where + T: Any + Send, + { + self.host_function_states + .remove(&TypeId::of::())? + .downcast::() + .ok() + .map(|state| *state) + } + + pub(crate) fn default_builtin_capabilities_enabled(&self) -> bool { + self.allow_default_builtin_capabilities } } diff --git a/src/vm/mod.rs b/src/vm/mod.rs index 37cd643c..96983ef3 100644 --- a/src/vm/mod.rs +++ b/src/vm/mod.rs @@ -24,8 +24,10 @@ mod superinstructions; mod tests; pub use self::aot::AotArtifactError; -pub use self::async_host::{CaptureAsyncHostContext, HostAsyncBridge, HostFuture}; -pub use self::capability::{CapabilityProfile, CapabilityProfileBuilder, IoPolicy}; +pub use self::async_host::{ + CaptureAsyncHostContext, HostAsyncBridge, HostFuture, HostFutureOutput, +}; +pub use self::capability::{CapabilityProfile, CapabilityProfileBuilder}; use self::engine::Engine; pub use self::epoch::{EpochCheckpoint, EpochHandle}; pub use self::fuel::FuelCheckpoint; @@ -40,48 +42,6 @@ use self::instance::{ExecutionFrame, FrameContinuation, Instance, QueuedCallable use self::run_context::{InterruptMode, RunContext}; pub use crate::builtins::runtime::cancellation::CancellationReason; -#[cfg(feature = "sqlite")] -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct SqliteLimits { - pub max_connections: usize, - pub max_statements: usize, - pub max_rows: usize, - pub max_columns: usize, - pub max_result_bytes: usize, - pub max_statement_bytes: usize, - pub max_parameters: usize, - pub max_parameter_bytes: usize, - pub max_pending_operations: usize, - pub max_transaction_ms: u64, - pub busy_timeout_ms: u64, -} - -#[cfg(feature = "sqlite")] -impl Default for SqliteLimits { - fn default() -> Self { - Self { - max_connections: 16, - max_statements: 128, - max_rows: 1_000, - max_columns: 128, - max_result_bytes: 4 * 1024 * 1024, - max_statement_bytes: 1024 * 1024, - max_parameters: 128, - max_parameter_bytes: 1024 * 1024, - max_pending_operations: 32, - max_transaction_ms: 5_000, - busy_timeout_ms: 5_000, - } - } -} - -#[cfg(feature = "sqlite")] -#[derive(Clone, Debug, Default, PartialEq, Eq)] -pub struct SqlitePolicy { - pub database_root: Option, - pub allow_unsafe_sql: bool, - pub limits: SqliteLimits, -} pub use crate::bytecode::{ CallableTarget, CallableValue, HostImport, OpCode, Program, Value, ValueType, }; diff --git a/src/vm/tests.rs b/src/vm/tests.rs index 8726ac37..f4790291 100644 --- a/src/vm/tests.rs +++ b/src/vm/tests.rs @@ -2,6 +2,8 @@ use super::async_host::WaitingHostOp; use super::*; use crate::builtins::BuiltinFunction; use crate::bytecode::TypeMap; +#[cfg(feature = "sqlite")] +use crate::{SqliteHostExt, SqlitePolicy}; use std::collections::HashMap; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; @@ -117,7 +119,9 @@ fn async_host_future_is_submitted_to_the_host_bridge() { })); let outcome = vm - .submit_host_future(Box::pin(async { Ok(CallReturn::one(Value::Int(42))) })) + .submit_host_future(Box::pin(async { + Ok(HostFutureOutput::returning(CallReturn::one(Value::Int(42)))) + })) .expect("host bridge should accept future"); let CallOutcome::Pending(op_id) = outcome else { panic!("async host submission should suspend"); @@ -132,7 +136,9 @@ fn async_host_future_is_submitted_to_the_host_bridge() { fn async_host_submission_without_driver_fails_and_retires_the_id() { let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); let error = vm - .submit_host_future(Box::pin(async { Ok(CallReturn::none()) })) + .submit_host_future(Box::pin(async { + Ok(HostFutureOutput::returning(CallReturn::none())) + })) .expect_err("missing host async driver should fail"); assert!( @@ -145,76 +151,112 @@ fn async_host_submission_without_driver_fails_and_retires_the_id() { } #[test] -fn unused_host_operation_ids_do_not_consume_registry_capacity() { - let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); - for _ in 0..128 { - vm.allocate_host_op_id(); +fn completing_a_submitted_host_op_cancels_the_driver_future() { + use std::sync::{Arc, Mutex}; + + struct CancelRecordingBridge(Arc>>); + + impl HostAsyncBridge for CancelRecordingBridge { + fn submit_op(&mut self, _op_id: HostOpId, _future: HostFuture) -> VmResult<()> { + Ok(()) + } + + fn poll_op( + &mut self, + _op_id: HostOpId, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Pending + } + + fn cancel_op(&mut self, op_id: HostOpId) { + self.0.lock().expect("cancel lock").push(op_id); + } } - assert_eq!(vm.host.runtime_operations.active_count(), 0); -} -#[cfg(feature = "async")] -#[test] -fn capability_profile_binding_installs_http_policy() { - let policy = crate::builtins::runtime::HttpConfig { - allowed_hosts: vec!["example.com".to_string()], - max_redirects: 2, - ..crate::builtins::runtime::HttpConfig::default() - }; - let mut registry = HostFunctionRegistry::empty(); - registry.set_capability_profile( - CapabilityProfile::builder() - .http_policy(policy.clone()) - .build(), - ); + let cancelled = Arc::new(Mutex::new(Vec::new())); let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); - registry - .bind_vm_cached(&mut vm) - .expect("profile should bind"); + vm.set_async_bridge(Box::new(CancelRecordingBridge(Arc::clone(&cancelled)))); + let CallOutcome::Pending(op_id) = vm + .submit_host_future(Box::pin(async { + Ok(HostFutureOutput::returning(CallReturn::none())) + })) + .expect("future should submit") + else { + panic!("submission should return pending"); + }; + vm.set_waiting_host_op(op_id) + .expect("submitted op should register"); - assert_eq!(vm.host.http_state.configuration(), Some(&policy)); + vm.complete_host_op(op_id, CallReturn::none()) + .expect("manual completion should succeed"); + + assert_eq!(*cancelled.lock().expect("cancel lock"), vec![op_id]); + assert_eq!(vm.waiting_host_op_id(), None); + assert_eq!(vm.host.runtime_operations.active_count(), 0); } -#[cfg(feature = "sqlite")] #[test] -fn capability_profile_binding_installs_sqlite_policy() { - let mut policy = crate::vm::SqlitePolicy::default(); - policy.limits.max_rows = 10; - let mut registry = HostFunctionRegistry::empty(); - registry.set_capability_profile( - CapabilityProfile::builder() - .sqlite_policy(policy.clone()) - .build(), - ); +fn failed_submitted_host_completion_clears_waiting_state() { + struct FailingCompletionBridge; + + impl HostAsyncBridge for FailingCompletionBridge { + fn submit_op(&mut self, _op_id: HostOpId, _future: HostFuture) -> VmResult<()> { + Ok(()) + } + + fn poll_op( + &mut self, + _op_id: HostOpId, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Pending + } + + fn poll_submitted_op( + &mut self, + _op_id: HostOpId, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(HostFutureOutput::complete(|_| { + Err(VmError::HostError("completion failed".to_string())) + }))) + } + } + let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); - registry - .bind_vm_cached(&mut vm) - .expect("profile should bind"); + vm.set_async_bridge(Box::new(FailingCompletionBridge)); + let CallOutcome::Pending(op_id) = vm + .submit_host_future(Box::pin(async { + Ok(HostFutureOutput::returning(CallReturn::none())) + })) + .expect("future should submit") + else { + panic!("submission should return pending"); + }; + vm.set_waiting_host_op(op_id) + .expect("submitted op should register"); + let waker = futures_util::task::noop_waker(); + let mut context = std::task::Context::from_waker(&waker); + + let result = vm.poll_waiting_host_op(&mut context); - assert_eq!(vm.host.sqlite_policy, policy); + assert!(matches!( + result, + std::task::Poll::Ready(Err(VmError::HostError(message))) + if message == "completion failed" + )); + assert_eq!(vm.waiting_host_op_id(), None); + assert_eq!(vm.host.runtime_operations.active_count(), 0); } #[test] -fn capability_profile_binding_installs_io_policy() { - let policy = crate::vm::IoPolicy { - allowed_roots: vec!["/tmp".to_string()], - allow_write: true, - allow_process: false, - max_read_bytes: 128, - max_write_bytes: 64, - }; - let mut registry = HostFunctionRegistry::empty(); - registry.set_capability_profile( - CapabilityProfile::builder() - .io_policy(policy.clone()) - .build(), - ); +fn unused_host_operation_ids_do_not_consume_registry_capacity() { let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); - registry - .bind_vm_cached(&mut vm) - .expect("profile should bind"); - - assert_eq!(vm.host.io_policy.as_ref(), Some(&policy)); + for _ in 0..128 { + vm.allocate_host_op_id(); + } + assert_eq!(vm.host.runtime_operations.active_count(), 0); } #[test] @@ -530,7 +572,7 @@ fn sqlite_reconfiguration_only_closes_sqlite_owned_state() { .expect("SQLite operation should start"); sqlite_operation.set_resource(sqlite_resource); - vm.configure_sqlite(crate::vm::SqlitePolicy::default()); + vm.configure_sqlite(SqlitePolicy::default()); assert!( vm.host diff --git a/tests/builtins/io_async_tests.rs b/tests/builtins/io_async_tests.rs new file mode 100644 index 00000000..f054dfb0 --- /dev/null +++ b/tests/builtins/io_async_tests.rs @@ -0,0 +1,106 @@ +use std::time::{SystemTime, UNIX_EPOCH}; + +use vm::{Value, Vm, VmError, VmStatus, compile_source}; + +fn run_source(source: &str) -> Result, VmError> { + let compiled = + compile_source(&format!("use io;\n{source}")).expect("async io source should compile"); + let mut vm = Vm::new(compiled.program); + super::async_test_bridge::install(&mut vm); + + let mut status = vm.run()?; + loop { + match status { + VmStatus::Halted => return Ok(vm.stack().to_vec()), + VmStatus::Yielded => status = vm.resume()?, + VmStatus::Waiting(_) => { + vm.wait_for_host_op_blocking()?; + status = vm.resume()?; + } + } + } +} + +#[test] +fn async_io_round_trips_file_operations_through_host_driver() { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock should follow Unix epoch") + .as_nanos(); + let path = std::env::temp_dir().join(format!("pd-vm-async-io-{}-{nonce}", std::process::id())); + + let stack = run_source(&format!( + r#" + let handle = io::open("{}", "w"); + io::write(handle, "host-driven"); + io::flush(handle); + io::close(handle); + io::exists("{}"); + "#, + path.display(), + path.display() + )) + .expect("async io program should complete"); + + assert_eq!(stack.last(), Some(&Value::Bool(true))); + assert_eq!( + std::fs::read_to_string(&path).expect("written file should exist"), + "host-driven" + ); + let _ = std::fs::remove_file(path); +} + +#[test] +fn async_io_read_line_preserves_buffered_data_between_calls() { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock should follow Unix epoch") + .as_nanos(); + let path = std::env::temp_dir().join(format!( + "pd-vm-async-read-line-{}-{nonce}", + std::process::id() + )); + std::fs::write(&path, "first\nsecond\n").expect("fixture should be written"); + + let stack = run_source(&format!( + r#" + let handle = io::open("{}", "r"); + io::read_line(handle); + let second = io::read_line(handle); + io::close(handle); + second; + "#, + path.display() + )) + .expect("async read_line program should complete"); + + assert_eq!(stack.last(), Some(&Value::string("second\n"))); + let _ = std::fs::remove_file(path); +} + +#[cfg(unix)] +#[test] +fn async_io_popen_reads_through_tokio_process_pipe() { + let stack = run_source( + r#" + let handle = io::popen("printf async-process", "r"); + let output = io::read_all(handle); + io::close(handle); + output; + "#, + ) + .expect("async popen program should complete"); + + assert_eq!(stack.last(), Some(&Value::string("async-process"))); +} + +#[test] +fn io_implementations_do_not_create_private_threads_or_runtimes() { + let async_source = include_str!("../../src/builtins/runtime/io/async_io.rs"); + let blocking_source = include_str!("../../src/builtins/runtime/io/blocking.rs"); + + assert!(!async_source.contains("thread::Builder")); + assert!(!async_source.contains("runtime::Builder")); + assert!(!async_source.contains("spawn_blocking")); + assert!(!blocking_source.contains("thread::Builder")); +} diff --git a/tests/builtins/io_builtin_edge_tests.rs b/tests/builtins/io_builtin_edge_tests.rs index f49283d6..b301fbea 100644 --- a/tests/builtins/io_builtin_edge_tests.rs +++ b/tests/builtins/io_builtin_edge_tests.rs @@ -1,6 +1,6 @@ use vm::{ - BuiltinFunction, CapabilityProfile, HostFunctionRegistry, IoPolicy, Value, Vm, VmError, - VmStatus, compile_source, + BuiltinFunction, CapabilityProfile, HostFunctionRegistry, IoHostExt, IoPolicy, Value, Vm, + VmError, VmStatus, compile_source, }; #[cfg(unix)] @@ -49,10 +49,10 @@ fn io_policy_denies_process_launch_when_process_capability_is_disabled() { registry.set_capability_profile( CapabilityProfile::builder() .allow_builtin(BuiltinFunction::IoPopen) - .io_policy(IoPolicy::default()) .build(), ); let mut vm = Vm::new(compiled.program); + vm.configure_io(IoPolicy::default()); registry .bind_vm_cached(&mut vm) .expect("profile should bind"); @@ -74,10 +74,10 @@ fn io_policy_denies_paths_outside_allowed_roots() { registry.set_capability_profile( CapabilityProfile::builder() .allow_builtin(BuiltinFunction::IoExists) - .io_policy(IoPolicy::default()) .build(), ); let mut vm = Vm::new(compiled.program); + vm.configure_io(IoPolicy::default()); registry .bind_vm_cached(&mut vm) .expect("profile should bind"); @@ -86,6 +86,32 @@ fn io_policy_denies_paths_outside_allowed_roots() { assert!(matches!(error, VmError::HostError(message) if message.contains("allowed roots"))); } +#[test] +fn restricted_registry_defaults_to_deny_when_io_host_state_is_absent() { + let compiled = compile_source( + r#" + use io; + io::exists("Cargo.toml"); + "#, + ) + .expect("source should compile"); + let mut registry = HostFunctionRegistry::restricted(); + registry.set_capability_profile( + CapabilityProfile::builder() + .allow_builtin(BuiltinFunction::IoExists) + .build(), + ); + let mut vm = Vm::new(compiled.program); + registry + .bind_vm_cached(&mut vm) + .expect("profile should bind"); + + let error = vm + .run() + .expect_err("missing IO host state should use the deny-by-default policy"); + assert!(matches!(error, VmError::HostError(message) if message.contains("allowed roots"))); +} + #[cfg(unix)] #[test] fn io_policy_limits_write_size() { @@ -110,10 +136,10 @@ fn io_policy_limits_write_size() { CapabilityProfile::builder() .allow_builtin(BuiltinFunction::IoOpen) .allow_builtin(BuiltinFunction::IoWrite) - .io_policy(policy) .build(), ); let mut vm = Vm::new(compiled.program); + vm.configure_io(policy); registry .bind_vm_cached(&mut vm) .expect("profile should bind"); @@ -153,10 +179,10 @@ fn io_policy_limits_read_all_size() { CapabilityProfile::builder() .allow_builtin(BuiltinFunction::IoOpen) .allow_builtin(BuiltinFunction::IoReadAll) - .io_policy(policy) .build(), ); let mut vm = Vm::new(compiled.program); + vm.configure_io(policy); registry .bind_vm_cached(&mut vm) .expect("profile should bind"); @@ -202,10 +228,10 @@ fn io_policy_limits_read_line_size() { CapabilityProfile::builder() .allow_builtin(BuiltinFunction::IoOpen) .allow_builtin(BuiltinFunction::IoReadLine) - .io_policy(policy) .build(), ); let mut vm = Vm::new(compiled.program); + vm.configure_io(policy); registry .bind_vm_cached(&mut vm) .expect("profile should bind"); @@ -244,8 +270,8 @@ fn process_exists(process_id: i32) -> bool { } #[test] -fn io_callback_resource_is_registered_before_worker_spawn() { - let source = include_str!("../../src/builtins/runtime/io.rs"); +fn blocking_io_runs_after_callback_registration_without_spawning_a_worker() { + let source = include_str!("../../src/builtins/runtime/io/blocking.rs"); let schedule = source .split_once("fn schedule_io_task(") .expect("schedule_io_task should exist") @@ -256,19 +282,14 @@ fn io_callback_resource_is_registered_before_worker_spawn() { let callback_registration = schedule .find(".insert(ResourceTypeId::CALLBACK, receiver)") .expect("schedule_io_task should register its callback receiver"); - let worker_spawn = schedule - .find(".spawn(move ||") - .expect("schedule_io_task should spawn its worker"); - assert!( - callback_registration < worker_spawn, - "callback receiver must be registered before the worker can run" - ); + assert!(!schedule.contains(".spawn(move ||")); + assert!(schedule[callback_registration..].contains("task()")); } #[test] fn popen_teardown_does_not_invoke_external_kill_programs() { - let source = include_str!("../../src/builtins/runtime/io.rs"); + let source = include_str!("../../src/builtins/runtime/io/blocking.rs"); assert!( !source.contains("Command::new(\"kill\")"), "Unix popen teardown must use the platform process API" @@ -290,8 +311,7 @@ fn reset_terminates_popen_descendants() { let compiled = compile_source(&format!( r#" use io; - let handle = io::popen("{command}", "r"); - io::read_all(handle); + io::popen("{command}", "r"); "# )) .expect("descendant popen source should compile"); @@ -301,8 +321,6 @@ fn reset_terminates_popen_descendants() { assert!(matches!(first, VmStatus::Waiting(_))); vm.wait_for_host_op_blocking() .expect("popen should complete"); - let second = vm.resume().expect("read_all should start"); - assert!(matches!(second, VmStatus::Waiting(_))); let pid_deadline = Instant::now() + Duration::from_secs(2); while !child_pid_path.exists() && Instant::now() < pid_deadline { @@ -330,6 +348,7 @@ fn reset_terminates_popen_descendants() { #[cfg(unix)] #[test] +#[ignore = "blocking IO runs the read on the caller thread"] fn reset_interrupts_a_blocked_popen_read_within_a_bounded_time() { let compiled = compile_source( r#" diff --git a/tests/builtins/stdlib_tests.rs b/tests/builtins/stdlib_tests.rs index cc1674f1..d6df463d 100644 --- a/tests/builtins/stdlib_tests.rs +++ b/tests/builtins/stdlib_tests.rs @@ -14,6 +14,8 @@ fn run_rustscript_spec(path: &Path) -> Vec { ); let mut vm = Vm::new(compiled.program); + #[cfg(feature = "async")] + super::async_test_bridge::install(&mut vm); loop { let status = vm.run().expect("spec vm should run"); match status { diff --git a/tests/builtins_tests.rs b/tests/builtins_tests.rs index e4f54996..e591e8a6 100644 --- a/tests/builtins_tests.rs +++ b/tests/builtins_tests.rs @@ -1,7 +1,16 @@ #![cfg(feature = "runtime")] +#[cfg(feature = "async")] +#[path = "support/async_test_bridge.rs"] +mod async_test_bridge; + +#[cfg(not(feature = "async"))] #[path = "builtins/io_builtin_edge_tests.rs"] mod io_builtin_edge_tests; +#[cfg(feature = "async")] +#[path = "builtins/io_async_tests.rs"] +mod io_async_tests; + #[path = "builtins/stdlib_tests.rs"] mod stdlib_tests; diff --git a/tests/compiler/compiler_rustscript_tests.rs b/tests/compiler/compiler_rustscript_tests.rs index 024a2bf1..b7fa4afc 100644 --- a/tests/compiler/compiler_rustscript_tests.rs +++ b/tests/compiler/compiler_rustscript_tests.rs @@ -163,6 +163,8 @@ fn rustscript_io_namespace_builtin_calls_are_supported() { "#; let compiled = compile_source(source).expect("compile should succeed"); let mut vm = Vm::new(compiled.program); + #[cfg(feature = "async")] + super::async_test_bridge::install(&mut vm); loop { let status = vm.run().expect("vm should run"); @@ -532,6 +534,8 @@ fn compile_source_file_with_rustscript_complex_fixture() { std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("examples/example_complex.rss"); let compiled = compile_source_file(path.as_path()).expect("compile should succeed"); let mut vm = Vm::new(compiled.program); + #[cfg(feature = "async")] + super::async_test_bridge::install(&mut vm); for func in &compiled.functions { match func.name.as_str() { diff --git a/tests/compiler_tests.rs b/tests/compiler_tests.rs index 11328072..d8ce5df8 100644 --- a/tests/compiler_tests.rs +++ b/tests/compiler_tests.rs @@ -1,5 +1,9 @@ #![allow(clippy::duplicate_mod)] +#[cfg(feature = "async")] +#[path = "support/async_test_bridge.rs"] +mod async_test_bridge; + #[cfg(feature = "runtime")] #[path = "compiler/compiler_common_tests.rs"] mod compiler_common_tests; diff --git a/tests/host_binding_generation_tests.rs b/tests/host_binding_generation_tests.rs index b663f106..2ce73ef4 100644 --- a/tests/host_binding_generation_tests.rs +++ b/tests/host_binding_generation_tests.rs @@ -311,78 +311,32 @@ fn capability_profile_fingerprint_uses_stable_callable_identities() { } #[test] -fn capability_profile_fingerprint_covers_http_policy() { - let first_policy = vm::HttpConfig { - allowed_hosts: vec!["example.com".to_string()], - max_redirects: 1, - ..vm::HttpConfig::default() - }; - let second_policy = vm::HttpConfig { - allowed_hosts: vec!["example.com".to_string()], - max_redirects: 2, - ..vm::HttpConfig::default() - }; - let first = CapabilityProfile::builder() - .http_policy(first_policy) - .build(); - let second = CapabilityProfile::builder() - .http_policy(second_policy) - .build(); - - assert_eq!(first.http_policy().expect("HTTP policy").max_redirects, 1); - assert_ne!(first.fingerprint(), second.fingerprint()); -} - -#[test] -fn capability_profile_fingerprint_covers_io_policy() { - let first = CapabilityProfile::builder() - .io_policy(vm::IoPolicy { - allowed_roots: vec!["/tmp/b".to_string(), "/tmp/a".to_string()], - max_read_bytes: 10, - ..vm::IoPolicy::default() - }) - .build(); - let reordered = CapabilityProfile::builder() - .io_policy(vm::IoPolicy { - allowed_roots: vec!["/tmp/a".to_string(), "/tmp/b".to_string()], - max_read_bytes: 10, - ..vm::IoPolicy::default() - }) - .build(); - let changed = CapabilityProfile::builder() - .io_policy(vm::IoPolicy { - allowed_roots: vec!["/tmp/a".to_string(), "/tmp/b".to_string()], - max_read_bytes: 11, - ..vm::IoPolicy::default() - }) - .build(); - - assert_eq!(first, reordered); - assert_eq!(first.fingerprint(), reordered.fingerprint()); - assert_ne!(first.fingerprint(), changed.fingerprint()); -} - -#[cfg(feature = "sqlite")] -#[test] -fn capability_profile_fingerprint_covers_sqlite_policy() { - let mut first_policy = vm::SqlitePolicy::default(); - first_policy.limits.max_rows = 10; - let mut second_policy = first_policy.clone(); - second_policy.limits.max_rows = 11; - let first = CapabilityProfile::builder() - .sqlite_policy(first_policy) - .build(); - let second = CapabilityProfile::builder() - .sqlite_policy(second_policy) - .build(); +fn vm_host_core_does_not_name_builtin_subsystem_policies() { + let manifest = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); + let host_runtime = std::fs::read_to_string(manifest.join("src/vm/host_runtime.rs")) + .expect("host runtime source"); + let capability = + std::fs::read_to_string(manifest.join("src/vm/capability.rs")).expect("capability source"); + let host = std::fs::read_to_string(manifest.join("src/vm/host.rs")).expect("host source"); - assert_eq!( - first - .sqlite_policy() - .expect("SQLite policy") - .limits - .max_rows, - 10 - ); - assert_ne!(first.fingerprint(), second.fingerprint()); + for forbidden in [ + "HttpState", + "IoPolicy", + "SqlitePolicy", + "http_state", + "io_policy", + "sqlite_policy", + ] { + assert!( + !host_runtime.contains(forbidden), + "HostRuntime leaked {forbidden}" + ); + assert!( + !capability.contains(forbidden), + "CapabilityProfile leaked {forbidden}" + ); + } + for forbidden in ["configure_http", "configure_sqlite", "http_is_configured"] { + assert!(!host.contains(forbidden), "Vm API leaked {forbidden}"); + } } diff --git a/tests/runtime_host_tests.rs b/tests/runtime_host_tests.rs index af5d5785..b81630f9 100644 --- a/tests/runtime_host_tests.rs +++ b/tests/runtime_host_tests.rs @@ -2,6 +2,8 @@ use std::sync::{Arc, Mutex}; +#[cfg(feature = "sqlite")] +use vm::SqliteHostExt; use vm::{ EventPayload, EventSink, HostFunctionRegistry, RuntimeResult, Value, Vm, VmStatus, compile_source, diff --git a/tests/support/async_test_bridge.rs b/tests/support/async_test_bridge.rs new file mode 100644 index 00000000..179abebd --- /dev/null +++ b/tests/support/async_test_bridge.rs @@ -0,0 +1,70 @@ +use std::collections::HashMap; +use std::task::{Context, Poll}; + +use vm::vm::{HostFuture, HostFutureOutput}; +use vm::{CallReturn, HostAsyncBridge, HostOpId, Vm, VmError, VmResult}; + +struct TokioTestBridge { + runtime: tokio::runtime::Runtime, + futures: HashMap, +} + +impl TokioTestBridge { + fn new() -> Self { + Self { + runtime: tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .expect("test runtime should build"), + futures: HashMap::new(), + } + } +} + +impl HostAsyncBridge for TokioTestBridge { + fn submit_op(&mut self, op_id: HostOpId, future: HostFuture) -> VmResult<()> { + if self.futures.insert(op_id, future).is_some() { + return Err(VmError::HostError(format!( + "duplicate submitted host op {op_id}" + ))); + } + Ok(()) + } + + fn poll_op(&mut self, op_id: HostOpId, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Err(VmError::HostError(format!( + "unexpected external op {op_id}" + )))) + } + + fn poll_submitted_op( + &mut self, + op_id: HostOpId, + cx: &mut Context<'_>, + ) -> Poll>> { + let poll = { + let future = match self.futures.get_mut(&op_id) { + Some(future) => future, + None => { + return Poll::Ready(Err(VmError::HostError(format!( + "unknown submitted host op {op_id}" + )))); + } + }; + let _guard = self.runtime.enter(); + future.as_mut().poll(cx) + }; + if poll.is_ready() { + self.futures.remove(&op_id); + } + poll + } + + fn cancel_op(&mut self, op_id: HostOpId) { + self.futures.remove(&op_id); + } +} + +pub(crate) fn install(vm: &mut Vm) { + vm.set_async_bridge(Box::new(TokioTestBridge::new())); +} diff --git a/tests/vm/http_host_tests.rs b/tests/vm/http_host_tests.rs index 9706da3c..531cd74c 100644 --- a/tests/vm/http_host_tests.rs +++ b/tests/vm/http_host_tests.rs @@ -1,12 +1,60 @@ +use std::collections::HashMap; use std::io::{Read, Write}; use std::net::TcpListener; +use std::task::{Context, Poll}; use std::thread; use vm::{ - CallOutcome, CallReturn, HostFunctionRegistry, HttpConfig, Program, Value, Vm, VmStatus, + CallOutcome, CallReturn, HostAsyncBridge, HostFunctionRegistry, HostFuture, HostFutureOutput, + HostOpId, HttpConfig, HttpHostExt, Program, Value, Vm, VmError, VmResult, VmStatus, compile_source, }; +#[derive(Default)] +struct TokioHostDriver { + submitted: HashMap, +} + +impl HostAsyncBridge for TokioHostDriver { + fn submit_op(&mut self, op_id: HostOpId, future: HostFuture) -> VmResult<()> { + self.submitted.insert(op_id, future); + Ok(()) + } + + fn poll_op(&mut self, op_id: HostOpId, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Err(VmError::HostError(format!( + "unknown external host operation {op_id}" + )))) + } + + fn poll_submitted_op( + &mut self, + op_id: HostOpId, + cx: &mut Context<'_>, + ) -> Poll> { + let poll = self.submitted.get_mut(&op_id).map_or_else( + || { + Poll::Ready(Err(VmError::HostError(format!( + "unknown submitted host operation {op_id}" + )))) + }, + |future| future.as_mut().poll(cx), + ); + if poll.is_ready() { + self.submitted.remove(&op_id); + } + poll + } + + fn cancel_op(&mut self, op_id: HostOpId) { + self.submitted.remove(&op_id); + } +} + +fn install_host_driver(vm: &mut Vm) { + vm.set_async_bridge(Box::::default()); +} + fn build_request_program(url: String) -> Program { compile_source(&format!( r#" @@ -85,6 +133,7 @@ async fn http_host_executes_a_bounded_request_and_returns_a_response_map() { let (port, server) = spawn_test_server(); let mut vm = Vm::new(build_request_program(format!("http://127.0.0.1:{port}/"))); vm.configure_http(local_http_config(port)); + install_host_driver(&mut vm); HostFunctionRegistry::new() .bind_vm_cached(&mut vm) .expect("default host registry should bind HTTP"); @@ -272,6 +321,7 @@ async fn explicitly_allowed_http_capability_reaches_http_policy() { allow_private_ips: true, ..HttpConfig::default() }); + install_host_driver(&mut vm); let mut registry = HostFunctionRegistry::restricted(); registry .allow_builtin("http::client::request") diff --git a/tests/vm/sqlite_host_tests.rs b/tests/vm/sqlite_host_tests.rs index 4551d827..20be44d3 100644 --- a/tests/vm/sqlite_host_tests.rs +++ b/tests/vm/sqlite_host_tests.rs @@ -1,7 +1,10 @@ extern crate vm as rustscript_vm; pub mod vm { + use std::any::{Any, TypeId}; + use std::collections::HashMap; + pub use crate::builtins::runtime::sqlite::{SqliteLimits, SqlitePolicy}; pub use crate::rustscript_vm::{ CallReturn, HostCallResult, HostOpId, OpCode, Program, Value, VmError, VmMap, VmResult, }; @@ -9,51 +12,32 @@ pub mod vm { use crate::builtins::runtime::cancellation::{CancellationToken, OperationRegistry}; use crate::builtins::runtime::resource::ResourceArena; - #[derive(Clone, Copy, Debug)] - pub struct SqliteLimits { - pub max_connections: usize, - pub max_statements: usize, - pub max_rows: usize, - pub max_columns: usize, - pub max_result_bytes: usize, - pub max_statement_bytes: usize, - pub max_parameters: usize, - pub max_parameter_bytes: usize, - pub max_pending_operations: usize, - pub max_transaction_ms: u64, - pub busy_timeout_ms: u64, + pub(crate) struct TestHostRuntime { + pub(crate) runtime_resources: ResourceArena, + pub(crate) runtime_operations: OperationRegistry, + host_function_states: HashMap>, } - impl Default for SqliteLimits { - fn default() -> Self { - Self { - max_connections: 16, - max_statements: 128, - max_rows: 1_000, - max_columns: 128, - max_result_bytes: 4 * 1024 * 1024, - max_statement_bytes: 1024 * 1024, - max_parameters: 128, - max_parameter_bytes: 1024 * 1024, - max_pending_operations: 32, - max_transaction_ms: 5_000, - busy_timeout_ms: 5_000, - } + impl TestHostRuntime { + pub(crate) fn set_host_function_state(&mut self, state: T) { + self.host_function_states + .insert(TypeId::of::(), Box::new(state)); } - } - - #[derive(Clone, Debug, Default)] - pub struct SqlitePolicy { - pub database_root: Option, - pub allow_unsafe_sql: bool, - pub limits: SqliteLimits, - } - pub(crate) struct TestHostRuntime { - pub(crate) runtime_resources: ResourceArena, - pub(crate) runtime_operations: OperationRegistry, + pub(crate) fn host_function_state(&self) -> Option<&T> { + self.host_function_states + .get(&TypeId::of::())? + .downcast_ref() + } - pub(crate) sqlite_policy: SqlitePolicy, + #[allow(dead_code)] + pub(crate) fn remove_host_function_state(&mut self) -> Option { + self.host_function_states + .remove(&TypeId::of::())? + .downcast::() + .ok() + .map(|state| *state) + } } pub(crate) struct TestRunContext { @@ -71,18 +55,13 @@ pub mod vm { host: TestHostRuntime { runtime_resources: ResourceArena::default(), runtime_operations: OperationRegistry::default(), - - sqlite_policy: SqlitePolicy::default(), + host_function_states: HashMap::new(), }, run_ctx: TestRunContext { cancellation: CancellationToken::root(), }, } } - - pub fn configure_sqlite(&mut self, policy: SqlitePolicy) { - self.host.sqlite_policy = policy; - } } } @@ -161,6 +140,28 @@ mod builtins { vm.host.runtime_resources.close(handle, reason) } + pub(crate) fn cancel_operations_by_owner( + vm: &mut crate::vm::Vm, + owner: cancellation::OperationOwner, + reason: cancellation::CancellationReason, + ) { + let operations = vm.host.runtime_operations.operations_by_owner(owner); + for operation in operations { + cancel_runtime_operation(vm, operation.id(), reason); + } + } + + pub(crate) fn close_resources_by_type( + vm: &mut crate::vm::Vm, + resource_type: resource::ResourceTypeId, + reason: cancellation::CancellationReason, + ) { + let handles = vm.host.runtime_resources.handles_of_type(resource_type); + for handle in handles { + let _ = close_runtime_resource(vm, handle, reason); + } + } + pub mod typed { pub type VmArrayRef<'a> = &'a [crate::vm::Value]; pub type VmMapRef<'a> = &'a crate::vm::VmMap; @@ -376,6 +377,7 @@ use std::sync::Arc; use std::task::{Context, Poll, Wake, Waker}; use std::time::{SystemTime, UNIX_EPOCH}; +use builtins::runtime::sqlite::SqliteHostExt; use builtins::runtime::test_api as sqlite; use vm::{CallReturn, HostCallResult, OpCode, Program, Value, Vm, VmError};