From 7ca7a27cd8b878eb4da10954cd3d7c334ae75dd4 Mon Sep 17 00:00:00 2001 From: fffonion Date: Mon, 17 Aug 2026 17:47:12 +0800 Subject: [PATCH] refactor(host): bind capabilities and async execution --- Cargo.lock | 49 ++ Cargo.toml | 2 + build.rs | 68 +- crates/rustscript/tests/alias_smoke.rs | 3 + pd-host-function/src/lib.rs | 258 +++++++- src/builtins/runtime/cancellation.rs | 3 + src/builtins/runtime/io/async_io.rs | 580 ++++++++++++++++++ .../runtime/{io.rs => io/blocking.rs} | 264 +++++--- src/builtins/runtime/io/mod.rs | 66 ++ src/builtins/runtime/mod.rs | 11 +- src/builtins/runtime/resource.rs | 5 + src/builtins/runtime/sqlite.rs | 102 ++- src/builtins/runtime/typed.rs | 29 + src/lib.rs | 14 +- src/vm/async_host/mod.rs | 361 +++++++++++ src/vm/capability.rs | 168 +++++ src/vm/host.rs | 407 +++--------- src/vm/host_runtime.rs | 58 +- src/vm/instance.rs | 2 +- src/vm/mod.rs | 53 +- src/vm/tests.rs | 173 +++++- tests/builtins/io_async_tests.rs | 106 ++++ tests/builtins/io_builtin_edge_tests.rs | 243 +++++++- tests/builtins/stdlib_tests.rs | 2 + tests/builtins_tests.rs | 9 + tests/common/mod.rs | 4 +- tests/compiler/compiler_rustscript_tests.rs | 4 + tests/compiler_tests.rs | 4 + tests/host_binding_generation_tests.rs | 84 ++- tests/runtime_host_tests.rs | 2 + tests/support/async_test_bridge.rs | 70 +++ tests/vm/sqlite_host_tests.rs | 94 +-- tests/vm/vm_runtime_tests.rs | 114 +++- 33 files changed, 2840 insertions(+), 572 deletions(-) create mode 100644 src/builtins/runtime/io/async_io.rs rename src/builtins/runtime/{io.rs => io/blocking.rs} (81%) create mode 100644 src/builtins/runtime/io/mod.rs create mode 100644 src/vm/async_host/mod.rs create mode 100644 src/vm/capability.rs create mode 100644 tests/builtins/io_async_tests.rs create mode 100644 tests/support/async_test_bridge.rs diff --git a/Cargo.lock b/Cargo.lock index c4430850..37764245 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -68,6 +68,12 @@ dependencies = [ "allocator-api2", ] +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + [[package]] name = "cc" version = "1.4.3" @@ -517,6 +523,17 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "mio" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + [[package]] name = "nibble_vec" version = "0.1.0" @@ -847,6 +864,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" @@ -859,6 +886,16 @@ version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -899,8 +936,14 @@ version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ + "bytes", + "libc", + "mio", "pin-project-lite", + "signal-hook-registry", + "socket2", "tokio-macros", + "windows-sys 0.61.2", ] [[package]] @@ -950,6 +993,12 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasmtime-internal-core" version = "42.0.2" diff --git a/Cargo.toml b/Cargo.toml index 7dfa5e63..87d8b99c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ name = "vm" [features] default = ["runtime", "cli", "cranelift-jit"] runtime = [] +async = ["runtime", "dep:tokio"] sqlite = ["runtime", "dep:rusqlite"] edge-abi = [ "dep:edge_abi", @@ -62,6 +63,7 @@ cranelift-module = { version = "0.129.1", optional = true } cranelift-native = { version = "0.129.1", optional = true } pd-host-function = { path = "./pd-host-function", version = "0.1.0" } rusqlite = { version = "0.32", default-features = false, features = ["bundled", "hooks", "limits"], 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 2da5cd0a..81a04201 100644 --- a/build.rs +++ b/build.rs @@ -115,6 +115,7 @@ struct CallableDecl { wrapper: Option, host_binding_kind: HostBindingKind, host_execution: HostExecutionKind, + runtime_owned_pending: bool, } #[derive(Clone, Debug)] @@ -243,10 +244,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() } @@ -268,6 +280,9 @@ fn parse_sources( } pub(crate) fn classify_host_binding(function: &ItemFn) -> HostBindingKind { + if function.sig.asyncness.is_some() { + return HostBindingKind::StaticStack; + } if function.sig.inputs.iter().any(|input| match input { FnArg::Typed(pat_type) => is_vm_context_type(&pat_type.ty), _ => false, @@ -293,6 +308,9 @@ pub(crate) fn classify_host_binding(function: &ItemFn) -> HostBindingKind { } pub(crate) fn infer_host_execution(function: &ItemFn) -> HostExecutionKind { + if function.sig.asyncness.is_some() { + return HostExecutionKind::MaySuspend; + } let return_type = normalized_return_type(&function.sig.output); if contains_host_call_result(&return_type) { HostExecutionKind::MaySuspend @@ -439,6 +457,8 @@ fn parse_source_file(path: &Path, spec: &SourceSpec, _order_offset: usize) -> Ve wrapper, host_binding_kind: classify_host_binding(function), host_execution: infer_host_execution(function), + runtime_owned_pending: function.sig.asyncness.is_none() + && contains_host_call_result(&normalized_return_type(&function.sig.output)), }); } out @@ -1090,12 +1110,14 @@ fn render_builtin_runtime_dispatch( ) .unwrap(); } - writeln!( - &mut out, - " registry.mark_runtime_owned_pending({:?});", - callable.name - ) - .unwrap(); + if callable.runtime_owned_pending { + writeln!( + &mut out, + " registry.mark_runtime_owned_pending({:?});", + callable.name + ) + .unwrap(); + } } writeln!(&mut out, "}}").unwrap(); writeln!(&mut out).unwrap(); @@ -1112,12 +1134,14 @@ fn render_builtin_runtime_dispatch( .render_bind_static_call(&callable.name, &host_wrapper_adapter_name(callable)); writeln!(&mut out, " {:?} => {{", callable.name).unwrap(); writeln!(&mut out, " {bind_call}").unwrap(); - writeln!( - &mut out, - " vm.mark_runtime_owned_pending_binding({:?});", - callable.name - ) - .unwrap(); + if callable.runtime_owned_pending { + writeln!( + &mut out, + " vm.mark_runtime_owned_pending_binding({:?});", + callable.name + ) + .unwrap(); + } writeln!(&mut out, " true").unwrap(); writeln!(&mut out, " }}").unwrap(); } @@ -1886,6 +1910,9 @@ fn host_wrapper_adapter_name(callable: &CallableDecl) -> String { fn generated_wrapper_decl(function: &ItemFn) -> WrapperDecl { let mut params = Vec::new(); + if function.sig.asyncness.is_some() { + params.push(WrapperParamKind::Vm); + } for input in &function.sig.inputs { let FnArg::Typed(pat_type) = input else { panic!("methods are not supported in #[pd_host_function] declarations"); @@ -1912,6 +1939,13 @@ fn parse_callable_params(function: &ItemFn) -> Vec { let FnArg::Typed(pat_type) = input else { panic!("methods are not supported in #[pd_host_function] declarations"); }; + if pat_type + .attrs + .iter() + .any(|attr| attr.path().is_ident("pd_host_context")) + { + return None; + } if is_vm_context_type(&pat_type.ty) { return None; } @@ -2078,7 +2112,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 8a649c0f..6dcb62a0 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 4aa1d3bc..fb4f8f96 100644 --- a/pd-host-function/src/lib.rs +++ b/pd-host-function/src/lib.rs @@ -8,7 +8,9 @@ use syn::{ #[proc_macro_attribute] pub fn pd_host_function(attr: TokenStream, item: TokenStream) -> TokenStream { let args = parse_macro_input!(attr with Punctuated::::parse_terminated); - match expand_pd_host_function(args, parse_macro_input!(item as ItemFn)) { + let item = parse_macro_input!(item as ItemFn); + let result = expand_pd_host_function(args, item); + match result { Ok(tokens) => tokens.into(), Err(err) => err.to_compile_error().into(), } @@ -19,9 +21,20 @@ fn expand_pd_host_function( mut item: ItemFn, ) -> Result { parse_name_arg(&attr)?; + let is_async = item.sig.asyncness.is_some(); let docs = doc_string(&item.attrs); for input in &item.sig.inputs { - validate_param(input)?; + if is_async { + validate_async_param(input)?; + } else if is_host_context_param(input) { + return Err(Error::new_spanned( + input, + "#[pd_host_context] is only valid on async host functions", + )); + } + if !is_host_context_param(input) { + validate_param(input)?; + } } validate_return_type(&item.sig.output)?; @@ -39,13 +52,85 @@ fn expand_pd_host_function( if item.sig.ident != impl_name { item.sig.ident = impl_name.clone(); } - let wrapper = generate_vm_wrapper(&item, &wrapper_name)?; + let wrapper = if is_async { + generate_async_vm_wrapper(&item, &wrapper_name)? + } else { + generate_vm_wrapper(&item, &wrapper_name)? + }; + for input in &mut item.sig.inputs { + if let FnArg::Typed(pat_type) = input { + pat_type + .attrs + .retain(|attr| !attr.path().is_ident("pd_host_context")); + } + } Ok(quote! { #item #wrapper }) } +fn validate_async_param(arg: &FnArg) -> Result<(), Error> { + let FnArg::Typed(pat_type) = arg else { + return Err(Error::new_spanned(arg, "methods are not supported")); + }; + if is_vm_context_type(&pat_type.ty) { + return Err(Error::new_spanned( + &pat_type.ty, + "async host functions cannot borrow Vm; capture owned host context before submission", + )); + } + if is_host_context_param(arg) { + return Ok(()); + } + if !is_async_owned_type(&pat_type.ty) { + return Err(Error::new_spanned( + &pat_type.ty, + "async host function parameters must be owned and 'static", + )); + } + Ok(()) +} + +fn is_host_context_param(arg: &FnArg) -> bool { + match arg { + FnArg::Typed(pat_type) => pat_type + .attrs + .iter() + .any(|attr| attr.path().is_ident("pd_host_context")), + FnArg::Receiver(_) => false, + } +} + +fn is_async_owned_type(ty: &Type) -> bool { + match ty { + Type::Group(group) => is_async_owned_type(&group.elem), + Type::Paren(paren) => is_async_owned_type(&paren.elem), + Type::Reference(_) | Type::Slice(_) => false, + Type::Tuple(tuple) => tuple.elems.iter().all(is_async_owned_type), + Type::Path(path) => { + let Some(segment) = path.path.segments.last() else { + return false; + }; + if matches!( + segment.ident.to_string().as_str(), + "str" | "VmStringRef" | "VmBytesRef" | "VmArrayRef" | "VmMapRef" | "VmValueRef" + ) { + return false; + } + match &segment.arguments { + syn::PathArguments::None => true, + syn::PathArguments::AngleBracketed(args) => args.args.iter().all(|arg| match arg { + syn::GenericArgument::Type(inner) => is_async_owned_type(inner), + _ => false, + }), + syn::PathArguments::Parenthesized(_) => false, + } + } + _ => false, + } +} + fn parse_name_arg(args: &Punctuated) -> Result { let Some(Meta::NameValue(name_value)) = args.first() else { return Err(Error::new( @@ -223,19 +308,110 @@ 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 } }) } +fn generate_async_vm_wrapper( + item: &ItemFn, + wrapper_name: &syn::Ident, +) -> Result { + let impl_name = &item.sig.ident; + let mutable_wrapper_name = syn::Ident::new(&format!("{wrapper_name}_mut"), wrapper_name.span()); + let mut extract_stmts = Vec::::new(); + let mut call_args = Vec::::new(); + let mut arg_index = 0usize; + + for input in &item.sig.inputs { + let FnArg::Typed(pat_type) = input else { + return Err(Error::new_spanned(input, "methods are not supported")); + }; + let Pat::Ident(PatIdent { ident, .. }) = pat_type.pat.as_ref() else { + return Err(Error::new_spanned( + &pat_type.pat, + "callable parameters must use identifier patterns", + )); + }; + let ty = &pat_type.ty; + if is_host_context_param(input) { + extract_stmts.push(quote! { + let #ident = <#ty as super::CaptureAsyncHostContext>::capture_with_args(vm, args)?; + }); + call_args.push(quote!(#ident)); + continue; + } + let label = LitStr::new( + &format!("{} {}", wrapper_name, ident), + proc_macro2::Span::call_site(), + ); + let index = syn::Index::from(arg_index); + extract_stmts.push(quote! { + let #ident = super::borrow_arg::<#ty>(args, #index, #label)?; + }); + call_args.push(quote!(#ident)); + arg_index += 1; + } + + let await_value = if return_is_vm_result(&item.sig.output) { + quote!(#impl_name(#(#call_args),*).await?) + } else { + quote!(#impl_name(#(#call_args),*).await) + }; + 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(super::HostFutureOutput::returning(values)) + } + super::CallOutcome::Pending(op_id) => Err(super::VmError::HostError( + format!("async host function returned nested pending operation {op_id}"), + )), + super::CallOutcome::Halt | super::CallOutcome::Yield => Err( + super::VmError::HostError( + "async host function returned a control-flow outcome".to_string(), + ), + ), + } + } + }; + 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(crate) fn #wrapper_name( + vm: &mut super::super::Vm, + args: &[super::super::Value], + ) -> super::super::VmResult { + #body + } + + #[allow(dead_code)] + pub(crate) fn #mutable_wrapper_name( + vm: &mut super::super::Vm, + args: &mut [super::super::Value], + ) -> super::super::VmResult { + #body + } + }) +} + fn wrapper_and_impl_names(name: &syn::Ident) -> (syn::Ident, syn::Ident) { let original = name.to_string(); match original.strip_suffix("_impl") { @@ -298,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") @@ -359,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, @@ -533,4 +723,60 @@ mod tests { .expect_err("the pd-host-function macro must not accept an async attribute"); assert!(error.to_string().contains("only supports name")); } + + #[test] + fn ordinary_async_signature_generates_host_driven_future_submission() { + let attr: Punctuated = parse_quote!(name = "test::async_call"); + let item: ItemFn = parse_quote!( + /// Returns an owned string asynchronously. + async fn async_call( + #[pd_host_context] context: TestContext, + value: String, + ) -> VmResult { + context.run(value).await + } + ); + let expanded = expand_pd_host_function(attr, item) + .expect("ordinary owned async function should use the generic async host contract") + .to_string(); + assert!(expanded.contains("submit_host_future")); + 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"); + let item: ItemFn = parse_quote! { + async fn borrowed(value: &str) -> VmResult { + Ok(value.to_string()) + } + }; + + let error = expand_pd_host_function(attr, item).expect_err("borrow should be rejected"); + assert!( + error + .to_string() + .contains("parameters must be owned and 'static") + ); + } } diff --git a/src/builtins/runtime/cancellation.rs b/src/builtins/runtime/cancellation.rs index 9e72aabf..108dbf92 100644 --- a/src/builtins/runtime/cancellation.rs +++ b/src/builtins/runtime/cancellation.rs @@ -409,6 +409,7 @@ impl OperationState { self.core.status() } + #[cfg_attr(feature = "async", allow(dead_code))] pub fn set_payload(&self, payload: ResourceHandle) { self.core .inner @@ -452,6 +453,7 @@ impl OperationState { .payload } + #[cfg_attr(feature = "async", allow(dead_code))] pub fn set_resource(&self, resource: ResourceHandle) { self.core .inner @@ -559,6 +561,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/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 81% rename from src/builtins/runtime/io.rs rename to src/builtins/runtime/io/blocking.rs index ee9e56cd..6c3bb0bf 100644 --- a/src/builtins/runtime/io.rs +++ b/src/builtins/runtime/io/blocking.rs @@ -1,6 +1,7 @@ use std::fs::OpenOptions; use std::future::Future; use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; use std::pin::Pin; use std::process::{Child, Command, Stdio}; use std::sync::atomic::{AtomicU32, Ordering}; @@ -14,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 }, @@ -134,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<'_>, @@ -167,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, @@ -189,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", )))) @@ -199,12 +205,21 @@ 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, ) -> VmResult> { - let path = path.to_string(); + let writes = match mode { + "r" => false, + "w" | "a" | "r+" | "w+" | "a+" => true, + other => { + return Err(VmError::HostError(format!( + "unsupported io_open mode '{other}', expected r/w/a/r+/w+/a+" + ))); + } + }; + let path = authorize_io_path(vm, path, writes)?; let mode = mode.to_string(); let op_id = schedule_io_task(vm, None, move || { let mut options = OpenOptions::new(); @@ -250,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, @@ -260,6 +275,11 @@ pub(super) fn builtin_io_popen( "unsupported io_popen mode '{mode}', expected r or w" ))); } + if super::io_policy(vm).is_some_and(|policy| !policy.allow_process) { + return Err(VmError::HostError( + "io_popen requires the process capability".to_string(), + )); + } let command = command.to_string(); let mode = mode.to_string(); let op_id = schedule_io_task(vm, None, move || { @@ -297,24 +317,28 @@ 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> { +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 || { let result = resource.with_handle_mut(|handle| { let mut out = String::new(); match handle { - IoHandle::File(file) => file - .read_to_string(&mut out) - .map_err(|err| VmError::HostError(format!("io_read_all failed: {err}")))?, - IoHandle::PopenRead { child } => child - .stdout - .as_mut() - .ok_or_else(|| { - VmError::HostError("io_read_all popen handle missing stdout".to_string()) - })? - .read_to_string(&mut out) - .map_err(|err| VmError::HostError(format!("io_read_all failed: {err}")))?, + IoHandle::File(file) => { + read_to_string_with_limit(file, max_read_bytes, &mut out)?; + } + IoHandle::PopenRead { child } => { + read_to_string_with_limit( + child.stdout.as_mut().ok_or_else(|| { + VmError::HostError( + "io_read_all popen handle missing stdout".to_string(), + ) + })?, + max_read_bytes, + &mut out, + )?; + } IoHandle::PopenWrite { .. } => { return Err(VmError::HostError( "io_read_all requires a readable handle".to_string(), @@ -330,21 +354,23 @@ pub(super) fn builtin_io_read_all(vm: &mut Vm, handle_id: i64) -> VmResult 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 || { let result = resource.with_handle_mut(|handle| { let line = match handle { - IoHandle::File(file) => read_line_from_reader(file)?, - IoHandle::PopenRead { child } => { - read_line_from_reader(child.stdout.as_mut().ok_or_else(|| { + IoHandle::File(file) => read_line_from_reader(file, max_read_bytes)?, + IoHandle::PopenRead { child } => read_line_from_reader( + child.stdout.as_mut().ok_or_else(|| { VmError::HostError("io_read_line popen handle missing stdout".to_string()) - })?)? - } + })?, + max_read_bytes, + )?, IoHandle::PopenWrite { .. } => { return Err(VmError::HostError( "io_read_line requires a readable handle".to_string(), @@ -360,11 +386,19 @@ 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) = super::io_policy(vm) + && text.len() > policy.max_write_bytes + { + return Err(VmError::HostError(format!( + "io_write exceeds the configured write limit of {} bytes", + policy.max_write_bytes + ))); + } let bytes = text.as_bytes().to_vec(); let handle = resource_handle(handle_id)?; let resource = io_resource_for_handle(vm, handle)?; @@ -397,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 || { @@ -425,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 || { @@ -444,16 +478,66 @@ pub(super) fn builtin_io_close(vm: &mut Vm, handle_id: i64) -> VmResult VmResult> { - let path = path.to_string(); +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( - std::path::Path::new(path.as_str()).exists(), - )))) + IoAsyncCompletion::result(Ok(CallReturn::one(Value::Bool(path.exists())))) })?; Ok(HostCallResult::Pending(op_id)) } +fn authorize_io_path(vm: &Vm, path: &str, writes: bool) -> VmResult { + let requested = PathBuf::from(path); + let Some(policy) = super::io_policy(vm) 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)?; + for root in &policy.allowed_roots { + let root = Path::new(root).canonicalize().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() + ))) +} + +fn canonicalize_io_target(path: &Path) -> VmResult { + if path.exists() { + return path + .canonicalize() + .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())) + })?; + parent + .canonicalize() + .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"); @@ -549,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) } @@ -747,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) => { @@ -882,7 +946,37 @@ fn terminate_process_tree(process_id: u32) -> VmResult<()> { ))) } -fn read_line_from_reader(reader: &mut impl Read) -> VmResult { +fn read_to_string_with_limit( + reader: &mut impl Read, + max_read_bytes: Option, + out: &mut String, +) -> VmResult<()> { + match max_read_bytes { + None => { + reader + .read_to_string(out) + .map_err(|err| VmError::HostError(format!("io_read_all failed: {err}")))?; + } + Some(limit) => { + let take_limit = u64::try_from(limit).unwrap_or(u64::MAX).saturating_add(1); + reader + .take(take_limit) + .read_to_string(out) + .map_err(|err| VmError::HostError(format!("io_read_all failed: {err}")))?; + if out.len() > limit { + return Err(VmError::HostError(format!( + "io_read_all exceeds the configured read limit of {limit} bytes" + ))); + } + } + } + Ok(()) +} + +fn read_line_from_reader( + reader: &mut impl Read, + max_read_bytes: Option, +) -> VmResult { let mut bytes = Vec::new(); let mut one = [0u8; 1]; loop { @@ -893,6 +987,12 @@ fn read_line_from_reader(reader: &mut impl Read) -> VmResult { break; } bytes.push(one[0]); + if max_read_bytes.is_some_and(|limit| bytes.len() > limit) { + return Err(VmError::HostError(format!( + "io_read_line exceeds the configured read limit of {} bytes", + max_read_bytes.expect("read limit should be present") + ))); + } if one[0] == b'\n' { break; } 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 97cfed06..8a4cf8ed 100644 --- a/src/builtins/runtime/mod.rs +++ b/src/builtins/runtime/mod.rs @@ -4,6 +4,8 @@ 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, HostFutureOutput, VmError}; use self::cancellation::{CancellationReason, OperationId, OperationOwner, OperationState}; use self::error::{RuntimeError, RuntimeErrorCode}; @@ -14,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), @@ -28,7 +31,6 @@ pub(crate) mod core; pub(crate) mod error; pub(crate) mod event; mod host; -#[cfg(not(target_arch = "wasm32"))] mod io; #[cfg(target_arch = "wasm32")] mod io_wasm; @@ -43,9 +45,9 @@ pub(crate) mod resource; mod sqlite; mod typed; -#[cfg(target_arch = "wasm32")] -use io_wasm as io; - +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, @@ -230,7 +232,6 @@ pub(crate) fn close_resources_by_type( } } -#[cfg(feature = "sqlite")] pub(crate) fn cancel_operations_by_owner( vm: &mut Vm, owner: OperationOwner, diff --git a/src/builtins/runtime/resource.rs b/src/builtins/runtime/resource.rs index 5281e8a8..49500e07 100644 --- a/src/builtins/runtime/resource.rs +++ b/src/builtins/runtime/resource.rs @@ -33,8 +33,10 @@ pub struct ResourceTypeId(u16); impl ResourceTypeId { pub const IO_FILE: Self = Self(1); + #[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 { @@ -183,6 +185,7 @@ impl ResourceArena { }) } + #[cfg_attr(feature = "async", allow(dead_code))] pub fn insert( &mut self, resource_type: ResourceTypeId, @@ -254,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, @@ -428,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 55612e0e..f54a8d88 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 { @@ -157,6 +166,15 @@ impl<'a> FromVmValue<'a> for &'a VmMap { } } +impl FromVmValue<'_> for VmMap { + fn from_vm_value(value: &Value, _label: &str) -> VmResult { + match value { + Value::Map(entries) => Ok(entries.as_ref().clone()), + _ => Err(VmError::TypeMismatch("map")), + } + } +} + impl FromVmValue<'_> for SharedArray { fn from_vm_value(value: &Value, _label: &str) -> VmResult { match value { @@ -453,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 ec4ea278..6ce3dbb1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -25,6 +25,10 @@ pub use assembler::{AsmParseError, Assembler, AssemblerError, BytecodeBuilder, a pub use builtins::runtime::HostCallResult; #[cfg(feature = "runtime")] pub use builtins::runtime::print::{PrintHostFunction, PrintlnHostFunction, format_value}; +#[cfg(feature = "runtime")] +pub use builtins::runtime::{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, @@ -86,15 +90,15 @@ pub use jit::{ pub use vm::diagnostics::render_vm_error; #[cfg(feature = "runtime")] pub use vm::{ - AotArtifactError, CallOutcome, CallReturn, CancellationReason, DEFAULT_MAX_SCRIPT_CALL_DEPTH, - EpochCheckpoint, EpochHandle, FuelCheckpoint, HostArgsFunction, HostAsyncBridge, - HostBindingPlan, HostFunction, HostFunctionRegistry, HostOpId, HostStackFunction, + AotArtifactError, CallOutcome, CallReturn, CancellationReason, CapabilityProfile, + CapabilityProfileBuilder, DEFAULT_MAX_SCRIPT_CALL_DEPTH, EpochCheckpoint, EpochHandle, + FuelCheckpoint, HostArgsFunction, HostAsyncBridge, HostBindingPlan, HostFunction, + 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 new file mode 100644 index 00000000..68ec8e80 --- /dev/null +++ b/src/vm/async_host/mod.rs @@ -0,0 +1,361 @@ +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll, Wake, Waker}; + +use super::*; + +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 { + fn submit_op(&mut self, _op_id: HostOpId, _future: HostFuture) -> VmResult<()> { + Err(VmError::HostError( + "async host bridge does not accept submitted futures".to_string(), + )) + } + + 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) { + self.cancel_op(op_id); + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct WaitingHostOp { + pub(super) op_id: HostOpId, +} + +struct NoopWake; + +impl Wake for NoopWake { + fn wake(self: Arc) {} +} + +fn noop_waker() -> Waker { + Waker::from(Arc::new(NoopWake)) +} + +impl Vm { + pub fn set_async_bridge(&mut self, bridge: Box) { + self.cancel_waiting_host_op(); + self.host.async_bridge = Some(bridge); + } + + pub fn clear_async_bridge(&mut self) { + self.cancel_waiting_host_op(); + self.host.async_bridge = None; + } + + pub fn allocate_host_op_id(&mut self) -> HostOpId { + self.host + .runtime_operations + .allocate_id() + .expect("host operation id space should not be exhausted") + .raw() + } + + pub fn submit_host_future(&mut self, future: HostFuture) -> VmResult { + let op_id = self.allocate_host_op_id(); + let bridge = self.host.async_bridge.as_mut().ok_or_else(|| { + 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)) + } + + pub fn waiting_host_op_id(&self) -> Option { + self.instance.waiting_host_op.map(|op| op.op_id) + } + + pub fn cancel_waiting_host_op(&mut self) { + self.cancel_waiting_host_op_with_reason( + crate::builtins::runtime::cancellation::CancellationReason::Requested, + ); + } + + pub(crate) fn cancel_waiting_host_op_with_reason( + &mut self, + reason: crate::builtins::runtime::cancellation::CancellationReason, + ) { + let Some(waiting) = self.instance.waiting_host_op.take() else { + return; + }; + let Ok(operation_id) = + crate::builtins::runtime::cancellation::OperationId::from_raw(waiting.op_id) + else { + return; + }; + let owner = self + .host + .runtime_operations + .get(operation_id) + .ok() + .map(|operation| operation.owner()); + if owner == Some(crate::builtins::runtime::cancellation::OperationOwner::HostBridge) { + 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); + } + } + + pub fn complete_host_op( + &mut self, + op_id: HostOpId, + values: impl Into, + ) -> VmResult<()> { + let waiting = self.instance.waiting_host_op.ok_or_else(|| { + VmError::HostError(format!( + "host op {op_id} completed but vm is not waiting on any op", + )) + })?; + if waiting.op_id != op_id { + return Err(VmError::HostError(format!( + "host op {op_id} completed while vm waits on {}", + waiting.op_id + ))); + } + let operation_id = crate::builtins::runtime::cancellation::OperationId::from_raw(op_id) + .map_err(|error| VmError::HostError(error.to_string()))?; + let operation = self + .host + .runtime_operations + .get(operation_id) + .map_err(|error| VmError::HostError(error.to_string()))?; + if operation.owner() != crate::builtins::runtime::cancellation::OperationOwner::HostBridge { + return Err(VmError::HostError(format!( + "host bridge cannot complete runtime-owned operation {op_id}", + ))); + } + self.host + .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()) + } + + pub fn poll_waiting_host_op(&mut self, cx: &mut Context<'_>) -> Poll> { + let Some(waiting) = self.instance.waiting_host_op else { + return Poll::Ready(Ok(())); + }; + let operation_id = + match crate::builtins::runtime::cancellation::OperationId::from_raw(waiting.op_id) { + Ok(operation_id) => operation_id, + Err(error) => return Poll::Ready(Err(VmError::HostError(error.to_string()))), + }; + let operation = match self.host.runtime_operations.get(operation_id) { + Ok(operation) => operation, + Err(error) => return Poll::Ready(Err(VmError::HostError(error.to_string()))), + }; + let host_bridge_owned = + operation.owner() == crate::builtins::runtime::cancellation::OperationOwner::HostBridge; + + let poll_result = if host_bridge_owned { + let bridge_ptr = match self.host.async_bridge.as_mut() { + Some(bridge) => bridge.as_mut() as *mut dyn HostAsyncBridge, + None => { + return Poll::Ready(Err(VmError::HostError(format!( + "vm waiting on host op {} without an async bridge", + waiting.op_id + )))); + } + }; + 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(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", + err.to_string(), + ) + .with_value(waiting.op_id); + let _ = self + .host + .runtime_operations + .fail(operation_id, runtime_error); + } + self.instance.waiting_host_op = None; + Poll::Ready(Err(err)) + } + } + } + + pub async fn await_waiting_host_op(&mut self) -> VmResult<()> { + std::future::poll_fn(|cx| self.poll_waiting_host_op(cx)).await + } + + pub fn wait_for_host_op_blocking(&mut self) -> VmResult<()> { + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + loop { + match self.poll_waiting_host_op(&mut cx) { + Poll::Ready(result) => return result, + Poll::Pending => { + #[cfg(not(target_arch = "wasm32"))] + { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + #[cfg(target_arch = "wasm32")] + { + return Err(VmError::HostError( + "blocking host-op wait is unsupported on wasm32 runtime".to_string(), + )); + } + } + } + } + } + + pub fn wait_for_host_op_blocking_with_cancel(&mut self, mut should_cancel: F) -> VmResult<()> + where + F: FnMut() -> bool, + { + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + loop { + if should_cancel() { + let cancellation_result = self + .run_ctx + .cancel(crate::builtins::runtime::cancellation::CancellationReason::Requested); + self.cancel_waiting_host_op(); + cancellation_result?; + return Err(VmError::HostError("host operation cancelled".to_string())); + } + match self.poll_waiting_host_op(&mut cx) { + Poll::Ready(result) => return result, + Poll::Pending => { + #[cfg(not(target_arch = "wasm32"))] + { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + #[cfg(target_arch = "wasm32")] + { + return Err(VmError::HostError( + "blocking host-op wait is unsupported on wasm32 runtime".to_string(), + )); + } + } + } + } + } +} diff --git a/src/vm/capability.rs b/src/vm/capability.rs new file mode 100644 index 00000000..9c60be11 --- /dev/null +++ b/src/vm/capability.rs @@ -0,0 +1,168 @@ +use crate::builtins::BuiltinFunction; + +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-v2"; + +/// Immutable authorization policy for privileged builtin calls and host imports. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CapabilityProfile { + allow_all_builtins: bool, + allow_all_host_imports: bool, + allowed_builtin_calls: Vec, + allowed_host_imports: Vec, + fingerprint: u64, +} + +impl CapabilityProfile { + pub fn builder() -> CapabilityProfileBuilder { + CapabilityProfileBuilder::default() + } + + pub fn deny_all() -> Self { + CapabilityProfileBuilder::default().build() + } + + pub fn allow_all() -> Self { + CapabilityProfileBuilder { + allow_all_builtins: true, + allow_all_host_imports: true, + ..CapabilityProfileBuilder::default() + } + .build() + } + + pub fn fingerprint(&self) -> u64 { + self.fingerprint + } + + pub fn allows_builtin(&self, builtin: BuiltinFunction) -> bool { + self.allow_all_builtins + || self + .allowed_builtin_calls + .binary_search(&builtin.call_index()) + .is_ok() + } + + pub fn allows_host_import(&self, name: &str) -> bool { + self.allow_all_host_imports + || self + .allowed_host_imports + .binary_search_by(|candidate| candidate.as_str().cmp(name)) + .is_ok() + } + + pub(crate) fn allowed_builtin_calls(&self) -> &[u16] { + &self.allowed_builtin_calls + } + + pub(crate) fn allows_all_builtins(&self) -> bool { + self.allow_all_builtins + } + + pub(crate) fn allows_all_host_imports(&self) -> bool { + self.allow_all_host_imports + } + + pub(crate) fn with_builtin(&self, builtin: BuiltinFunction) -> Self { + let mut builder = CapabilityProfileBuilder { + allow_all_builtins: self.allow_all_builtins, + 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(), + }; + builder.allowed_builtin_calls.push(builtin.call_index()); + builder.build() + } + + pub(crate) fn with_host_import(&self, name: &str) -> Self { + let mut builder = CapabilityProfileBuilder { + allow_all_builtins: self.allow_all_builtins, + 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(), + }; + builder.allowed_host_imports.push(name.to_string()); + builder.build() + } +} + +impl Default for CapabilityProfile { + fn default() -> Self { + Self::deny_all() + } +} + +#[derive(Clone, Debug, Default)] +pub struct CapabilityProfileBuilder { + allow_all_builtins: bool, + allow_all_host_imports: bool, + allowed_builtin_calls: Vec, + allowed_host_imports: Vec, +} + +impl CapabilityProfileBuilder { + pub fn allow_builtin(mut self, builtin: BuiltinFunction) -> Self { + self.allowed_builtin_calls.push(builtin.call_index()); + self + } + + pub fn allow_host_import(mut self, name: impl Into) -> Self { + self.allowed_host_imports.push(name.into()); + 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(); + let fingerprint = fingerprint( + self.allow_all_builtins, + self.allow_all_host_imports, + &self.allowed_builtin_calls, + &self.allowed_host_imports, + ); + 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, + fingerprint, + } + } +} + +fn fingerprint( + allow_all_builtins: bool, + allow_all_host_imports: bool, + builtin_calls: &[u16], + host_imports: &[String], +) -> u64 { + let mut value = FNV_OFFSET_BASIS; + update_fingerprint(&mut value, PROFILE_VERSION); + update_fingerprint( + &mut value, + &[ + u8::from(allow_all_builtins), + u8::from(allow_all_host_imports), + ], + ); + update_fingerprint(&mut value, &(builtin_calls.len() as u64).to_le_bytes()); + for call in builtin_calls { + update_fingerprint(&mut value, &call.to_le_bytes()); + } + update_fingerprint(&mut value, &(host_imports.len() as u64).to_le_bytes()); + for name in host_imports { + update_fingerprint(&mut value, &(name.len() as u64).to_le_bytes()); + update_fingerprint(&mut value, name.as_bytes()); + } + value +} + +fn update_fingerprint(state: &mut u64, bytes: &[u8]) { + for byte in bytes { + *state ^= u64::from(*byte); + *state = state.wrapping_mul(FNV_PRIME); + } +} diff --git a/src/vm/host.rs b/src/vm/host.rs index 3d73b53e..676997b4 100644 --- a/src/vm/host.rs +++ b/src/vm/host.rs @@ -1,9 +1,9 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, OnceLock, RwLock}; -use std::task::{Context, Poll, Wake, Waker}; use crate::builtins::BuiltinFunction; +use super::async_host::WaitingHostOp; use super::*; pub type HostOpId = u64; @@ -86,16 +86,6 @@ pub trait HostArgsFunction: Send { fn call(&mut self, args: &[Value]) -> VmResult; } -pub trait HostAsyncBridge: Send { - fn poll_op(&mut self, op_id: HostOpId, cx: &mut Context<'_>) -> Poll>; - - fn cancel_op(&mut self, _op_id: HostOpId) {} - - fn cancel_op_with_reason(&mut self, op_id: HostOpId, _reason: CancellationReason) { - self.cancel_op(op_id); - } -} - pub type StaticHostFunction = fn(&mut Vm, &[Value]) -> VmResult; pub type StaticHostStackFunction = fn(&mut Vm, &[Value]) -> VmResult; pub type StaticHostArgsFunction = fn(&[Value]) -> VmResult; @@ -132,8 +122,8 @@ pub struct HostBindingPlan { allow_default_builtin_capabilities: bool, allowed_host_function_slots: Vec, allow_default_host_capabilities: bool, - capability_profile: Arc<()>, - capability_state: Arc<()>, + capability_profile: Arc, + capability_fingerprint: u64, registry_state: Arc<()>, registry_generation_token: Arc<()>, registry_generation: u64, @@ -146,10 +136,8 @@ pub struct HostFunctionRegistry { plan_cache: Arc, Arc>>>, allowed_builtin_calls: Arc>, allow_default_builtin_capabilities: bool, - allowed_host_registry_slots: Arc>, allow_default_host_capabilities: bool, - capability_profile: Arc<()>, - capability_state: Arc<()>, + capability_profile: Arc, registry_state: Arc<()>, registry_generation_token: Arc<()>, registry_generation: Arc, @@ -169,10 +157,8 @@ impl HostFunctionRegistry { plan_cache: Arc::new(RwLock::new(HashMap::new())), allowed_builtin_calls: Arc::new(Vec::new()), allow_default_builtin_capabilities: true, - allowed_host_registry_slots: Arc::new(Vec::new()), allow_default_host_capabilities: true, - capability_profile: Arc::new(()), - capability_state: Arc::new(()), + capability_profile: Arc::new(CapabilityProfile::allow_all()), registry_state: Arc::new(()), registry_generation_token: Arc::new(()), registry_generation: Arc::new(AtomicU64::new(0)), @@ -192,8 +178,7 @@ impl HostFunctionRegistry { }) .clone(); registry.plan_cache = Arc::new(RwLock::new(HashMap::new())); - registry.capability_profile = Arc::new(()); - registry.capability_state = Arc::new(()); + registry.capability_profile = Arc::new(CapabilityProfile::allow_all()); registry.registry_state = Arc::new(()); registry.registry_generation_token = Arc::new(()); registry.registry_generation = Arc::new(AtomicU64::new(0)); @@ -206,9 +191,7 @@ impl HostFunctionRegistry { let mut registry = Self::new(); registry.allow_default_builtin_capabilities = false; registry.allow_default_host_capabilities = false; - registry.allowed_host_registry_slots = Arc::new(Vec::new()); - registry.capability_profile = Arc::new(()); - registry.capability_state = Arc::new(()); + registry.capability_profile = Arc::new(CapabilityProfile::deny_all()); registry.registry_state = Arc::new(()); registry.registry_generation_token = Arc::new(()); registry.registry_generation = Arc::new(AtomicU64::new(0)); @@ -216,16 +199,20 @@ impl HostFunctionRegistry { registry } + /// Replaces the registry's immutable capability profile. + pub fn set_capability_profile(&mut self, profile: CapabilityProfile) { + self.allowed_builtin_calls = Arc::new(profile.allowed_builtin_calls().to_vec()); + self.allow_default_builtin_capabilities = profile.allows_all_builtins(); + self.allow_default_host_capabilities = profile.allows_all_host_imports(); + self.capability_profile = Arc::new(profile); + self.invalidate_plan_cache(); + } + /// Explicitly permits a namespaced builtin when this registry is used as a capability plan. pub fn allow_builtin(&mut self, name: impl AsRef) -> VmResult<()> { let name = name.as_ref(); - if let Some(®istry_slot) = self.by_name.get(name) { - let slots = Arc::make_mut(&mut self.allowed_host_registry_slots); - if !slots.contains(®istry_slot) { - slots.push(registry_slot); - slots.sort_unstable(); - } - self.capability_state = Arc::new(()); + if self.by_name.contains_key(name) { + self.capability_profile = Arc::new(self.capability_profile.with_host_import(name)); self.invalidate_plan_cache(); return Ok(()); } @@ -236,7 +223,7 @@ impl HostFunctionRegistry { calls.push(builtin.call_index()); calls.sort_unstable(); } - self.capability_state = Arc::new(()); + self.capability_profile = Arc::new(self.capability_profile.with_builtin(builtin)); self.invalidate_plan_cache(); Ok(()) } @@ -247,6 +234,7 @@ impl HostFunctionRegistry { self.plan_cache = Arc::new(RwLock::new(HashMap::new())); } + #[allow(dead_code)] pub(crate) fn mark_runtime_owned_pending(&mut self, name: &str) { let slot = self .by_name @@ -455,7 +443,51 @@ impl HostFunctionRegistry { self.invalidate_plan_cache(); } + fn validate_builtin_capability(&self, call_index: u16) -> VmResult<()> { + if let Some(builtin) = BuiltinFunction::from_call_index(call_index) + && builtin.requires_explicit_host_capability() + && !self.allowed_builtin_calls.contains(&call_index) + { + return Err(VmError::HostError(format!( + "capability profile does not allow builtin '{}'", + builtin.name() + ))); + } + Ok(()) + } + + fn validate_program_capabilities(&self, program: &Program) -> VmResult<()> { + if self.allow_default_builtin_capabilities { + return Ok(()); + } + let mut ip = 0usize; + while let Some(&raw_opcode) = program.code.get(ip) { + let opcode = + OpCode::try_from(raw_opcode).map_err(|_| VmError::InvalidOpcode(raw_opcode))?; + let operand_end = ip + .checked_add(1 + opcode.operand_len()) + .ok_or(VmError::BytecodeBounds)?; + if operand_end > program.code.len() { + return Err(VmError::BytecodeBounds); + } + if opcode == OpCode::Call { + let bytes: [u8; 2] = program.code[ip + 1..ip + 3] + .try_into() + .map_err(|_| VmError::BytecodeBounds)?; + self.validate_builtin_capability(u16::from_le_bytes(bytes))?; + } + ip = operand_end; + } + for prototype in &program.callable_prototypes { + if let CallableTarget::HostImport(call_index) = prototype.target { + self.validate_builtin_capability(call_index)?; + } + } + Ok(()) + } + pub fn bind_vm_cached(&self, vm: &mut Vm) -> VmResult<()> { + self.validate_program_capabilities(&vm.program)?; let plan = self.prepare_shared_plan(&vm.program.imports)?; self.bind_vm_with_plan(vm, &plan) } @@ -469,8 +501,8 @@ impl HostFunctionRegistry { } fn plan_matches_current(&self, plan: &HostBindingPlan) -> bool { - Arc::ptr_eq(&self.capability_profile, &plan.capability_profile) - && Arc::ptr_eq(&self.capability_state, &plan.capability_state) + self.capability_profile.fingerprint() == plan.capability_fingerprint + && self.capability_profile.as_ref() == plan.capability_profile.as_ref() && Arc::ptr_eq(&self.registry_state, &plan.registry_state) && Arc::ptr_eq( &self.registry_generation_token, @@ -505,6 +537,14 @@ impl HostFunctionRegistry { .entries .get(registry_slot as usize) .ok_or(VmError::InvalidCall(registry_slot))?; + if !self.allow_default_host_capabilities + && !self.capability_profile.allows_host_import(&import.name) + { + return Err(VmError::HostError(format!( + "capability profile does not allow host import '{}'", + import.name + ))); + } if entry.arity != import.arity { return Err(VmError::InvalidCallArity { import: import.name.clone(), @@ -524,16 +564,17 @@ impl HostFunctionRegistry { resolved_calls.push(vm_slot); } - let allowed_host_function_slots = self - .allowed_host_registry_slots + let mut allowed_host_function_slots = imports .iter() - .filter_map(|registry_slot| { - registry_slots - .iter() - .position(|slot| slot == registry_slot) - .map(|slot| slot as u16) + .zip(resolved_calls.iter().copied()) + .filter_map(|(import, vm_slot)| { + self.capability_profile + .allows_host_import(&import.name) + .then_some(vm_slot) }) - .collect(); + .collect::>(); + allowed_host_function_slots.sort_unstable(); + allowed_host_function_slots.dedup(); let runtime_owned_pending_slots = registry_slots .iter() .enumerate() @@ -555,7 +596,7 @@ impl HostFunctionRegistry { allowed_host_function_slots, allow_default_host_capabilities: self.allow_default_host_capabilities, capability_profile: Arc::clone(&self.capability_profile), - capability_state: Arc::clone(&self.capability_state), + capability_fingerprint: self.capability_profile.fingerprint(), registry_state: Arc::clone(&self.registry_state), registry_generation_token: Arc::clone(&self.registry_generation_token), registry_generation: self.registry_generation.load(Ordering::Relaxed), @@ -569,21 +610,19 @@ impl HostFunctionRegistry { } pub fn bind_vm_with_plan(&self, vm: &mut Vm, plan: &HostBindingPlan) -> VmResult<()> { + self.validate_program_capabilities(&vm.program)?; if vm.program.imports != plan.import_signature { return Err(VmError::HostError( "host binding plan does not match vm import signature".to_string(), )); } - if !Arc::ptr_eq(&self.capability_profile, &plan.capability_profile) { + if self.capability_profile.fingerprint() != plan.capability_fingerprint + || self.capability_profile.as_ref() != plan.capability_profile.as_ref() + { return Err(VmError::HostError( "host binding plan belongs to a different capability profile".to_string(), )); } - if !Arc::ptr_eq(&self.capability_state, &plan.capability_state) { - return Err(VmError::HostError( - "host binding plan belongs to a different capability state".to_string(), - )); - } if !Arc::ptr_eq(&self.registry_state, &plan.registry_state) { return Err(VmError::HostError( "host binding plan belongs to a different registry state".to_string(), @@ -716,21 +755,6 @@ pub(crate) fn validate_non_yielding_host_value( Err(VmError::TypeMismatch(expected)) } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(super) struct WaitingHostOp { - pub(super) op_id: HostOpId, -} - -struct NoopWake; - -impl Wake for NoopWake { - fn wake(self: Arc) {} -} - -fn noop_waker() -> Waker { - Waker::from(Arc::new(NoopWake)) -} - #[inline] fn builtin_for_binding_name(name: &str) -> Option { if !name.contains("::") { @@ -826,6 +850,7 @@ impl Vm { } } + #[allow(dead_code)] pub(crate) fn mark_runtime_owned_pending_binding(&mut self, name: &str) { let slot = builtin_for_binding_name(name) .and_then(|builtin| { @@ -1056,16 +1081,6 @@ impl Vm { .insert(builtin_call_index, host_slot); } - pub fn set_async_bridge(&mut self, bridge: Box) { - self.cancel_waiting_host_op(); - self.host.async_bridge = Some(bridge); - } - - pub fn clear_async_bridge(&mut self) { - self.cancel_waiting_host_op(); - self.host.async_bridge = None; - } - pub fn set_runtime_print_sink(&mut self, sink: F) where F: FnMut(String) + Send + 'static, @@ -1131,36 +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(); - } - /// Enables or disables implicit binding of built-in host functions. /// /// Disabling this makes the VM use only explicitly registered host functions. The default @@ -1184,208 +1169,6 @@ impl Vm { Ok(()) } - pub fn allocate_host_op_id(&mut self) -> HostOpId { - self.host - .runtime_operations - .allocate_id() - .expect("host operation id space should not be exhausted") - .raw() - } - - pub fn waiting_host_op_id(&self) -> Option { - self.instance.waiting_host_op.map(|op| op.op_id) - } - - pub fn cancel_waiting_host_op(&mut self) { - self.cancel_waiting_host_op_with_reason( - crate::builtins::runtime::cancellation::CancellationReason::Requested, - ); - } - - pub(crate) fn cancel_waiting_host_op_with_reason( - &mut self, - reason: crate::builtins::runtime::cancellation::CancellationReason, - ) { - let Some(waiting) = self.instance.waiting_host_op.take() else { - return; - }; - let Ok(operation_id) = - crate::builtins::runtime::cancellation::OperationId::from_raw(waiting.op_id) - else { - return; - }; - let owner = self - .host - .runtime_operations - .get(operation_id) - .ok() - .map(|operation| operation.owner()); - if owner == Some(crate::builtins::runtime::cancellation::OperationOwner::HostBridge) { - if let Some(bridge) = self.host.async_bridge.as_mut() { - bridge.cancel_op_with_reason(waiting.op_id, reason); - } - let _ = self.host.runtime_operations.cancel(operation_id, reason); - } else { - crate::builtins::runtime::cancel_builtin_io_op_with_reason(self, waiting.op_id, reason); - } - } - - pub fn complete_host_op( - &mut self, - op_id: HostOpId, - values: impl Into, - ) -> VmResult<()> { - let waiting = self.instance.waiting_host_op.ok_or_else(|| { - VmError::HostError(format!( - "host op {op_id} completed but vm is not waiting on any op", - )) - })?; - if waiting.op_id != op_id { - return Err(VmError::HostError(format!( - "host op {op_id} completed while vm waits on {}", - waiting.op_id - ))); - } - let operation_id = crate::builtins::runtime::cancellation::OperationId::from_raw(op_id) - .map_err(|error| VmError::HostError(error.to_string()))?; - let operation = self - .host - .runtime_operations - .get(operation_id) - .map_err(|error| VmError::HostError(error.to_string()))?; - if operation.owner() != crate::builtins::runtime::cancellation::OperationOwner::HostBridge { - return Err(VmError::HostError(format!( - "host bridge cannot complete runtime-owned operation {op_id}", - ))); - } - self.host - .runtime_operations - .complete(operation_id) - .map_err(|error| VmError::HostError(error.to_string()))?; - self.complete_waiting_host_op(op_id, values.into()) - } - - pub fn poll_waiting_host_op(&mut self, cx: &mut Context<'_>) -> Poll> { - let Some(waiting) = self.instance.waiting_host_op else { - return Poll::Ready(Ok(())); - }; - let operation_id = - match crate::builtins::runtime::cancellation::OperationId::from_raw(waiting.op_id) { - Ok(operation_id) => operation_id, - Err(error) => return Poll::Ready(Err(VmError::HostError(error.to_string()))), - }; - let operation = match self.host.runtime_operations.get(operation_id) { - Ok(operation) => operation, - Err(error) => return Poll::Ready(Err(VmError::HostError(error.to_string()))), - }; - let host_bridge_owned = - operation.owner() == crate::builtins::runtime::cancellation::OperationOwner::HostBridge; - - let poll_result = if host_bridge_owned { - let bridge_ptr = match self.host.async_bridge.as_mut() { - Some(bridge) => bridge.as_mut() as *mut dyn HostAsyncBridge, - None => { - return Poll::Ready(Err(VmError::HostError(format!( - "vm waiting on host op {} without an async bridge", - waiting.op_id - )))); - } - }; - unsafe { (&mut *bridge_ptr).poll_op(waiting.op_id, cx) } - } else { - crate::builtins::runtime::poll_builtin_io_op(self, waiting.op_id, cx) - }; - - match poll_result { - Poll::Pending => Poll::Pending, - Poll::Ready(Ok(values)) => { - if host_bridge_owned { - self.host - .runtime_operations - .complete(operation_id) - .map_err(|error| VmError::HostError(error.to_string()))?; - } - self.complete_waiting_host_op(waiting.op_id, values)?; - Poll::Ready(Ok(())) - } - Poll::Ready(Err(err)) => { - if host_bridge_owned { - 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; - Poll::Ready(Err(err)) - } - } - } - - pub async fn await_waiting_host_op(&mut self) -> VmResult<()> { - std::future::poll_fn(|cx| self.poll_waiting_host_op(cx)).await - } - - pub fn wait_for_host_op_blocking(&mut self) -> VmResult<()> { - let waker = noop_waker(); - let mut cx = Context::from_waker(&waker); - loop { - match self.poll_waiting_host_op(&mut cx) { - Poll::Ready(result) => return result, - Poll::Pending => { - #[cfg(not(target_arch = "wasm32"))] - { - std::thread::sleep(std::time::Duration::from_millis(1)); - } - #[cfg(target_arch = "wasm32")] - { - return Err(VmError::HostError( - "blocking host-op wait is unsupported on wasm32 runtime".to_string(), - )); - } - } - } - } - } - - pub fn wait_for_host_op_blocking_with_cancel(&mut self, mut should_cancel: F) -> VmResult<()> - where - F: FnMut() -> bool, - { - let waker = noop_waker(); - let mut cx = Context::from_waker(&waker); - loop { - if should_cancel() { - let cancellation_result = self - .run_ctx - .cancel(crate::builtins::runtime::cancellation::CancellationReason::Requested); - self.cancel_waiting_host_op(); - cancellation_result?; - return Err(VmError::HostError("host operation cancelled".to_string())); - } - match self.poll_waiting_host_op(&mut cx) { - Poll::Ready(result) => return result, - Poll::Pending => { - #[cfg(not(target_arch = "wasm32"))] - { - std::thread::sleep(std::time::Duration::from_millis(1)); - } - #[cfg(target_arch = "wasm32")] - { - return Err(VmError::HostError( - "blocking host-op wait is unsupported on wasm32 runtime".to_string(), - )); - } - } - } - } - } - pub(super) fn execute_host_call( &mut self, index: u16, @@ -1532,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 8782bbd3..c53b4d61 100644 --- a/src/vm/host_runtime.rs +++ b/src/vm/host_runtime.rs @@ -3,14 +3,16 @@ //! [`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/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::cancellation::{ @@ -18,9 +20,8 @@ use crate::builtins::runtime::cancellation::{ }; use crate::builtins::runtime::resource::{DEFAULT_MAX_RESOURCES, ResourceArena}; -#[cfg(feature = "sqlite")] -use crate::vm::SqlitePolicy; -use crate::vm::host::{HostAsyncBridge, VmHostFunction}; +use crate::vm::async_host::HostAsyncBridge; +use crate::vm::host::VmHostFunction; /// Embedder-supplied print sink for `print`/`debug` output. pub(crate) type RuntimePrintSink = dyn FnMut(String) + Send; @@ -45,9 +46,9 @@ pub(crate) struct HostRuntime { pub(crate) resolved_calls_dirty: bool, pub(crate) runtime_resources: ResourceArena, pub(crate) runtime_operations: OperationRegistry, - #[cfg(feature = "sqlite")] - pub(crate) sqlite_policy: SqlitePolicy, + host_function_states: HashMap>, pub(crate) async_bridge: Option>, + pub(crate) submitted_host_ops: HashSet, pub(crate) runtime_print_sink: Option>, } @@ -71,9 +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"), - #[cfg(feature = "sqlite")] - sqlite_policy: SqlitePolicy::default(), + host_function_states: HashMap::new(), async_bridge: None, + submitted_host_ops: HashSet::new(), runtime_print_sink: None, } } @@ -89,6 +90,39 @@ impl HostRuntime { let _ = self .runtime_resources .close_all(CancellationReason::VmReset); + 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 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/instance.rs b/src/vm/instance.rs index baecdfb0..475ff791 100644 --- a/src/vm/instance.rs +++ b/src/vm/instance.rs @@ -18,7 +18,7 @@ use std::sync::atomic::AtomicBool; use std::sync::{Arc, Weak}; use crate::bytecode::{CallableValue, Program, SharedCaptureCell, Value}; -use crate::vm::host::WaitingHostOp; +use crate::vm::async_host::WaitingHostOp; use crate::vm::map_iter::MapIteratorState; use crate::vm::{DEFAULT_MAX_SCRIPT_CALL_DEPTH, VmYieldReason}; diff --git a/src/vm/mod.rs b/src/vm/mod.rs index 0709c6cd..96983ef3 100644 --- a/src/vm/mod.rs +++ b/src/vm/mod.rs @@ -4,6 +4,8 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; pub(crate) mod aot; +mod async_host; +mod capability; pub mod diagnostics; mod engine; mod epoch; @@ -21,12 +23,17 @@ mod superinstructions; #[cfg(test)] mod tests; pub use self::aot::AotArtifactError; + +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; pub use self::host::{ - CallOutcome, CallReturn, HostArgsFunction, HostAsyncBridge, HostBindingPlan, HostFunction, - HostFunctionRegistry, HostOpId, HostStackFunction, StaticHostArgsFunction, StaticHostFunction, + CallOutcome, CallReturn, HostArgsFunction, HostBindingPlan, HostFunction, HostFunctionRegistry, + HostOpId, HostStackFunction, StaticHostArgsFunction, StaticHostFunction, StaticHostStackFunction, }; use self::host::{HostCallExecOutcome, VmHostFunction}; @@ -35,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)] -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)] -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 057949f9..35f44031 100644 --- a/src/vm/tests.rs +++ b/src/vm/tests.rs @@ -1,7 +1,9 @@ -use super::host::WaitingHostOp; +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}; @@ -27,6 +29,7 @@ fn failed_dynamic_builtin_override_preserves_runtime_owned_pending_binding() { vm.ensure_call_bindings() .expect("default fallback should bind runtime sleep"); let slot = vm.host.host_function_symbols["runtime::sleep"]; + vm.host.runtime_owned_pending_host_slots.insert(slot); assert!(vm.host.runtime_owned_pending_host_slots.contains(&slot)); vm.bind_builtin_override("runtime::sleep", Box::new(Dummy)) @@ -47,6 +50,7 @@ fn failed_static_builtin_override_preserves_runtime_owned_pending_binding() { vm.ensure_call_bindings() .expect("default fallback should bind runtime sleep"); let slot = vm.host.host_function_symbols["runtime::sleep"]; + vm.host.runtime_owned_pending_host_slots.insert(slot); assert!(vm.host.runtime_owned_pending_host_slots.contains(&slot)); vm.bind_builtin_static_override("runtime::sleep", dummy) @@ -81,6 +85,171 @@ fn reset_for_reuse_keeps_host_operation_ids_monotonic() { assert_eq!(vm.allocate_host_op_id(), 2); } +#[test] +fn async_host_future_is_submitted_to_the_host_bridge() { + use std::sync::{Arc, Mutex}; + + struct RecordingBridge { + submitted: Arc>>, + future: Arc>>, + } + + impl HostAsyncBridge for RecordingBridge { + fn submit_op(&mut self, op_id: HostOpId, future: HostFuture) -> VmResult<()> { + self.submitted.lock().expect("submitted lock").push(op_id); + *self.future.lock().expect("future lock") = Some(future); + Ok(()) + } + + fn poll_op( + &mut self, + _op_id: HostOpId, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Pending + } + } + + let submitted = Arc::new(Mutex::new(Vec::new())); + let future = Arc::new(Mutex::new(None)); + let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); + vm.set_async_bridge(Box::new(RecordingBridge { + submitted: Arc::clone(&submitted), + future: Arc::clone(&future), + })); + + let outcome = vm + .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"); + }; + + assert_eq!(*submitted.lock().expect("submitted lock"), vec![op_id]); + assert!(future.lock().expect("future lock").is_some()); + assert_eq!(vm.host.runtime_operations.active_count(), 0); +} + +#[test] +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(HostFutureOutput::returning(CallReturn::none())) + })) + .expect_err("missing host async driver should fail"); + + assert!( + error + .to_string() + .contains("async host function requires a host async bridge") + ); + assert_eq!(vm.allocate_host_op_id(), 2); + assert_eq!(vm.host.runtime_operations.active_count(), 0); +} + +#[test] +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); + } + } + + let cancelled = Arc::new(Mutex::new(Vec::new())); + let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); + 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"); + + 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); +} + +#[test] +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])); + 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!(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 unused_host_operation_ids_do_not_consume_registry_capacity() { let mut vm = Vm::new(Program::new(Vec::new(), vec![OpCode::Ret as u8])); @@ -403,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 9348ed44..b301fbea 100644 --- a/tests/builtins/io_builtin_edge_tests.rs +++ b/tests/builtins/io_builtin_edge_tests.rs @@ -1,4 +1,7 @@ -use vm::{Value, Vm, VmError, VmStatus, compile_source}; +use vm::{ + BuiltinFunction, CapabilityProfile, HostFunctionRegistry, IoHostExt, IoPolicy, Value, Vm, + VmError, VmStatus, compile_source, +}; #[cfg(unix)] use std::path::PathBuf; @@ -33,6 +36,223 @@ fn run_source_host_error(source: &str) -> String { } } +#[test] +fn io_policy_denies_process_launch_when_process_capability_is_disabled() { + let compiled = compile_source( + r#" + use io; + io::popen("exit 0", "r"); + "#, + ) + .expect("source should compile"); + let mut registry = HostFunctionRegistry::restricted(); + registry.set_capability_profile( + CapabilityProfile::builder() + .allow_builtin(BuiltinFunction::IoPopen) + .build(), + ); + let mut vm = Vm::new(compiled.program); + vm.configure_io(IoPolicy::default()); + registry + .bind_vm_cached(&mut vm) + .expect("profile should bind"); + + let error = vm.run().expect_err("process launch should be denied"); + assert!(matches!(error, VmError::HostError(message) if message.contains("process capability"))); +} + +#[test] +fn io_policy_denies_paths_outside_allowed_roots() { + 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); + vm.configure_io(IoPolicy::default()); + registry + .bind_vm_cached(&mut vm) + .expect("profile should bind"); + + let error = vm.run().expect_err("path should be denied"); + 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() { + let path = unique_temp_path("policy-write-limit"); + let compiled = compile_source(&format!( + r#" + use io; + let handle = io::open("{}", "w"); + io::write(handle, "four"); + "#, + path.display() + )) + .expect("source should compile"); + let policy = IoPolicy { + allowed_roots: vec![std::env::temp_dir().display().to_string()], + allow_write: true, + max_write_bytes: 3, + ..IoPolicy::default() + }; + let mut registry = HostFunctionRegistry::restricted(); + registry.set_capability_profile( + CapabilityProfile::builder() + .allow_builtin(BuiltinFunction::IoOpen) + .allow_builtin(BuiltinFunction::IoWrite) + .build(), + ); + let mut vm = Vm::new(compiled.program); + vm.configure_io(policy); + registry + .bind_vm_cached(&mut vm) + .expect("profile should bind"); + + assert!(matches!( + vm.run().expect("open should start"), + VmStatus::Waiting(_) + )); + vm.wait_for_host_op_blocking() + .expect("open should complete"); + let error = vm.resume().expect_err("oversized write should be denied"); + assert!(matches!(error, VmError::HostError(message) if message.contains("write limit"))); + let _ = std::fs::remove_file(path); +} + +#[cfg(unix)] +#[test] +fn io_policy_limits_read_all_size() { + let path = unique_temp_path("policy-read-limit"); + std::fs::write(&path, "four").expect("fixture should be written"); + let compiled = compile_source(&format!( + r#" + use io; + let handle = io::open("{}", "r"); + io::read_all(handle); + "#, + path.display() + )) + .expect("source should compile"); + let policy = IoPolicy { + allowed_roots: vec![std::env::temp_dir().display().to_string()], + max_read_bytes: 3, + ..IoPolicy::default() + }; + let mut registry = HostFunctionRegistry::restricted(); + registry.set_capability_profile( + CapabilityProfile::builder() + .allow_builtin(BuiltinFunction::IoOpen) + .allow_builtin(BuiltinFunction::IoReadAll) + .build(), + ); + let mut vm = Vm::new(compiled.program); + vm.configure_io(policy); + registry + .bind_vm_cached(&mut vm) + .expect("profile should bind"); + + assert!(matches!( + vm.run().expect("open should start"), + VmStatus::Waiting(_) + )); + vm.wait_for_host_op_blocking() + .expect("open should complete"); + assert!(matches!( + vm.resume().expect("read should start"), + VmStatus::Waiting(_) + )); + let error = vm + .wait_for_host_op_blocking() + .expect_err("oversized read should be denied"); + assert!(matches!(error, VmError::HostError(message) if message.contains("read limit"))); + let _ = std::fs::remove_file(path); +} + +#[cfg(unix)] +#[test] +fn io_policy_limits_read_line_size() { + let path = unique_temp_path("policy-read-line-limit"); + std::fs::write(&path, "four\n").expect("fixture should be written"); + let compiled = compile_source(&format!( + r#" + use io; + let handle = io::open("{}", "r"); + io::read_line(handle); + "#, + path.display() + )) + .expect("source should compile"); + let policy = IoPolicy { + allowed_roots: vec![std::env::temp_dir().display().to_string()], + max_read_bytes: 3, + ..IoPolicy::default() + }; + let mut registry = HostFunctionRegistry::restricted(); + registry.set_capability_profile( + CapabilityProfile::builder() + .allow_builtin(BuiltinFunction::IoOpen) + .allow_builtin(BuiltinFunction::IoReadLine) + .build(), + ); + let mut vm = Vm::new(compiled.program); + vm.configure_io(policy); + registry + .bind_vm_cached(&mut vm) + .expect("profile should bind"); + + assert!(matches!( + vm.run().expect("open should start"), + VmStatus::Waiting(_) + )); + vm.wait_for_host_op_blocking() + .expect("open should complete"); + assert!(matches!( + vm.resume().expect("read should start"), + VmStatus::Waiting(_) + )); + let error = vm + .wait_for_host_op_blocking() + .expect_err("oversized line should be denied"); + assert!(matches!(error, VmError::HostError(message) if message.contains("read limit"))); + let _ = std::fs::remove_file(path); +} + #[cfg(unix)] fn unique_temp_path(label: &str) -> PathBuf { let nonce = SystemTime::now() @@ -50,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") @@ -62,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" @@ -96,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"); @@ -107,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 { @@ -136,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/common/mod.rs b/tests/common/mod.rs index 89fc659d..335a8d64 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -1,8 +1,8 @@ #![allow(unused_imports)] pub use vm::{ - Assembler, BytecodeBuilder, CallOutcome, CompileSourceFileOptions, Compiler, Expr, - HostArgsFunction, HostFunction, HostFunctionRegistry, Program, SourceFlavor, + Assembler, BytecodeBuilder, CallOutcome, CapabilityProfile, CompileSourceFileOptions, Compiler, + Expr, HostArgsFunction, HostFunction, HostFunctionRegistry, Program, SourceFlavor, StaticHostArgsFunction, Stmt, Store, Value, Vm, VmStatus, assemble, compile_source, compile_source_file, compile_source_file_with_options, compile_source_with_flavor, }; 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 3e44c1af..2ce73ef4 100644 --- a/tests/host_binding_generation_tests.rs +++ b/tests/host_binding_generation_tests.rs @@ -6,7 +6,10 @@ use build_script::{ HostBindingKind, HostExecutionKind, classify_host_binding, infer_host_execution, }; use syn::parse_quote; -use vm::{HostFunctionRegistry, JitConfig, JitTraceTerminal, Value, Vm, VmStatus, compile_source}; +use vm::{ + BuiltinFunction, CapabilityProfile, HostFunctionRegistry, JitConfig, JitTraceTerminal, Value, + Vm, VmStatus, compile_source, +}; fn native_jit_supported() -> bool { (cfg!(target_arch = "x86_64") @@ -136,6 +139,18 @@ fn infers_host_suspension_from_the_return_signature() { ); } + let asynchronous = parse_quote!( + async fn host(value: String) -> VmResult {} + ); + assert_eq!( + infer_host_execution(&asynchronous), + HostExecutionKind::MaySuspend + ); + assert_eq!( + classify_host_binding(&asynchronous), + HostBindingKind::StaticStack + ); + let synchronous = parse_quote!( fn host() -> VmResult {} ); @@ -230,11 +245,15 @@ fn restricted_capabilities_disable_trace_jit_for_host_imports_and_builtins() { hot_loop_threshold: 1, max_trace_len: 512, }); - HostFunctionRegistry::restricted() + let error = HostFunctionRegistry::restricted() .bind_vm_cached(&mut vm) - .expect("restricted registry should bind program imports"); + .expect_err("restricted registry should reject ungranted capability during preflight"); - assert!(matches!(vm.run(), Err(vm::VmError::UnboundImport(_)))); + assert!( + error + .to_string() + .contains("capability profile does not allow") + ); assert_eq!(vm.jit_native_exec_count(), 0); } } @@ -264,3 +283,60 @@ fn runtime_exit_still_halts_for_direct_and_cached_default_bindings() { assert!(vm.stack().is_empty()); } } + +#[test] +fn capability_profile_fingerprint_uses_stable_callable_identities() { + let first = CapabilityProfile::builder() + .allow_builtin(BuiltinFunction::JsonEncode) + .allow_host_import("custom::echo") + .build(); + let reordered = CapabilityProfile::builder() + .allow_host_import("custom::echo") + .allow_builtin(BuiltinFunction::JsonEncode) + .build(); + + assert_eq!(first, reordered); + assert_eq!(first.fingerprint(), reordered.fingerprint()); + assert!(first.allows_builtin(BuiltinFunction::JsonEncode)); + assert!(first.allows_host_import("custom::echo")); + assert!(!first.allows_host_import("custom::other")); + assert_ne!( + first.fingerprint(), + CapabilityProfile::deny_all().fingerprint() + ); + assert_ne!( + CapabilityProfile::allow_all().fingerprint(), + CapabilityProfile::deny_all().fingerprint() + ); +} + +#[test] +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"); + + 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/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}; diff --git a/tests/vm/vm_runtime_tests.rs b/tests/vm/vm_runtime_tests.rs index 8fa22ef1..92536a81 100644 --- a/tests/vm/vm_runtime_tests.rs +++ b/tests/vm/vm_runtime_tests.rs @@ -45,6 +45,39 @@ fn empty_registry_allows_functions_registered_by_the_embedder() { assert_eq!(vm.stack(), &[Value::Int(42)]); } +#[test] +fn explicit_capability_profile_authorizes_host_imports_during_preflight() { + let program = compile_source("fn action() -> int; action();") + .expect("host call source should compile") + .program; + let mut registry = HostFunctionRegistry::empty(); + registry.register_static_args("action", 0, returns_registered_value); + registry.set_capability_profile(CapabilityProfile::deny_all()); + + let mut denied = Vm::new(program.clone()); + let error = registry + .bind_vm_cached(&mut denied) + .expect_err("deny-all profile must reject the host import during binding"); + assert!(error.to_string().contains("capability")); + + let mut allowed_registry = HostFunctionRegistry::empty(); + allowed_registry.set_capability_profile( + CapabilityProfile::builder() + .allow_host_import("action") + .build(), + ); + allowed_registry.register_static_args("action", 0, returns_registered_value); + let mut allowed = Vm::new(program); + allowed_registry + .bind_vm_cached(&mut allowed) + .expect("allowed host import should bind"); + assert_eq!( + allowed.run().expect("host call should run"), + VmStatus::Halted + ); + assert_eq!(allowed.stack(), &[Value::Int(42)]); +} + #[test] fn empty_registry_preserves_default_builtin_capabilities() { let compiled = compile_source("use bytes; bytes::from_array_u8([1, 2, 3]);") @@ -58,29 +91,81 @@ fn empty_registry_preserves_default_builtin_capabilities() { assert_eq!(vm.stack(), &[Value::bytes(vec![1, 2, 3])]); } +#[test] +fn explicit_capability_profile_authorizes_builtin_calls_during_preflight() { + let program = compile_source("use bytes; bytes::from_array_u8([1, 2, 3]);") + .expect("bytes source should compile") + .program; + let mut registry = HostFunctionRegistry::empty(); + registry.set_capability_profile(CapabilityProfile::deny_all()); + + let mut denied = Vm::new(program.clone()); + let error = registry + .bind_vm_cached(&mut denied) + .expect_err("deny-all profile must reject builtin calls during binding"); + assert!(error.to_string().contains("capability")); + + registry.set_capability_profile( + CapabilityProfile::builder() + .allow_builtin(vm::BuiltinFunction::BytesFromArrayU8) + .build(), + ); + let mut allowed = Vm::new(program); + registry + .bind_vm_cached(&mut allowed) + .expect("allowed builtin should bind"); + assert_eq!( + allowed.run().expect("builtin call should run"), + VmStatus::Halted + ); + assert_eq!(allowed.stack(), &[Value::bytes(vec![1, 2, 3])]); +} + +#[test] +fn explicit_capability_profile_rejects_builtin_callable_metadata_during_preflight() { + let mut program = Program::new(Vec::new(), vec![OpCode::Ret as u8]); + program.callable_prototypes.push(vm::CallablePrototype { + kind: vm::CallableKind::HostFunction, + target: vm::CallableTarget::HostImport(vm::BuiltinFunction::BytesFromArrayU8.call_index()), + arity: 1, + frame_local_count: 0, + parameter_slots: Vec::new(), + capture_source_slots: Vec::new(), + capture_slots: Vec::new(), + capture_modes: Vec::new(), + self_slot: None, + schema: None, + }); + let mut vm = Vm::new(program); + let mut registry = HostFunctionRegistry::empty(); + registry.set_capability_profile(CapabilityProfile::deny_all()); + + let error = registry + .bind_vm_cached(&mut vm) + .expect_err("builtin callable metadata must be authorized during binding"); + assert!(error.to_string().contains("capability")); +} + #[cfg(feature = "cranelift-jit")] #[test] -fn restricted_builtin_capabilities_match_between_interpreter_and_aot() { +fn restricted_builtin_capabilities_are_rejected_before_interpreter_or_aot_execution() { let source = "use bytes; bytes::from_array_u8([1, 2, 3]);"; let program = compile_source(source) .expect("bytes source should compile") .program; let mut interpreter = Vm::new(program.clone()); - HostFunctionRegistry::restricted() + let interpreter_error = HostFunctionRegistry::restricted() .bind_vm_cached(&mut interpreter) - .expect("restricted registry should bind"); - assert!(matches!( - interpreter.run(), - Err(vm::VmError::UnboundImport(_)) - )); + .expect_err("restricted profile should reject before interpreter execution"); let mut aot = Vm::new(program); - HostFunctionRegistry::restricted() - .bind_vm_cached(&mut aot) - .expect("restricted registry should bind"); aot.compile_aot().expect("AOT compile should succeed"); - assert!(matches!(aot.run(), Err(vm::VmError::UnboundImport(_)))); + let aot_error = HostFunctionRegistry::restricted() + .bind_vm_cached(&mut aot) + .expect_err("restricted profile should reject before AOT execution"); + assert_eq!(interpreter_error.to_string(), aot_error.to_string()); + assert!(interpreter_error.to_string().contains("capability")); } #[test] @@ -404,11 +489,10 @@ fn builtin_override_does_not_bypass_restricted_capability_profile() { .program; let mut denied = Vm::new(program.clone()); - HostFunctionRegistry::restricted() + let error = HostFunctionRegistry::restricted() .bind_vm_cached(&mut denied) - .expect("restricted registry should bind"); - denied.bind_function("io::exists", Box::new(ExistsOverride)); - assert!(matches!(denied.run(), Err(vm::VmError::UnboundImport(_)))); + .expect_err("restricted profile should reject before override installation"); + assert!(error.to_string().contains("capability")); let mut allowed_registry = HostFunctionRegistry::restricted(); allowed_registry