From bdcd3dc7f494ff85b6be261a771dcd2fb535a8bc Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Mon, 13 Jul 2026 21:50:16 +0200 Subject: [PATCH] =?UTF-8?q?fix(executor):=20bound=20Rhai=E2=86=92JSON=20ma?= =?UTF-8?q?terialization=20to=20prevent=20anonymous=20OOM?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-element Rhai sandbox caps (max_string_size 64 KiB, max_array_size / max_map_size 10 000) do NOT bound the *materialized* size: Rhai shares strings and arrays by Rc, so a 10 000-element array of one aliased 64 KiB string is cheap to build (~10 000 ops, far under the 1 M op budget) yet dealiases to ~640 MiB of distinct JSON. Every Dynamic→JSON conversion deep-copies the aliases; the HTTP response body path had NO size cap at all, and the KV/docs value caps run only AFTER full materialization — so a handful of anonymous requests could OOM the node (32 concurrent × ~640 MiB ≈ 20 GiB) on the stated consumer-hardware target. Add a byte-budgeted materializer that bails the moment the budget is exceeded, so the transient allocation is bounded regardless of aliasing: - bridge.rs: `dynamic_to_json_capped(value, max) -> Result` (charges a running budget) + `MAX_JSON_MATERIALIZE_BYTES` (16 MiB hard rail, far above the 256 KiB business caps). `dynamic_to_json` stays infallible for best-effort paths (logging) but is now internally bounded — it collapses an over-limit value to a marker string instead of OOMing. - Route every user-value boundary through the fail-closed capped form: KV set/set_if (+ the CAS `expected`), docs create/update/find, secrets set, http request body, workflow input, `json::stringify`, and the HTTP RESPONSE body (previously the one fully-uncapped exit → now a 500). `invoke` args and the pubsub/queue message materializers get the same byte budget in place. - `impl From for Box` so SDK sites protect themselves with a bare `?`. Pinned by bridge unit tests: an aliased 10 000×64 KiB array errors on the budget rather than materializing, and the infallible wrapper yields a marker instead of OOMing. Workspace 914 + journeys 157/157 green. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/executor-core/src/engine.rs | 16 +- crates/executor-core/src/sdk/bridge.rs | 164 ++++++++++++++++++-- crates/executor-core/src/sdk/docs.rs | 20 +-- crates/executor-core/src/sdk/http.rs | 8 +- crates/executor-core/src/sdk/invoke.rs | 59 +++++-- crates/executor-core/src/sdk/kv.rs | 25 +-- crates/executor-core/src/sdk/mod.rs | 5 +- crates/executor-core/src/sdk/pubsub.rs | 73 +++++++-- crates/executor-core/src/sdk/queue.rs | 57 +++++-- crates/executor-core/src/sdk/secrets.rs | 6 +- crates/executor-core/src/sdk/stdlib/json.rs | 6 +- crates/executor-core/src/sdk/workflow.rs | 4 +- 12 files changed, 363 insertions(+), 80 deletions(-) diff --git a/crates/executor-core/src/engine.rs b/crates/executor-core/src/engine.rs index bcf48d0..24365ff 100644 --- a/crates/executor-core/src/engine.rs +++ b/crates/executor-core/src/engine.rs @@ -57,7 +57,9 @@ use crate::module_resolver::{ }; use crate::sandbox::Limits; use crate::sdk; -use crate::sdk::bridge::{dynamic_to_json, json_to_dynamic}; +use crate::sdk::bridge::{ + dynamic_to_json, dynamic_to_json_capped, json_to_dynamic, MAX_JSON_MATERIALIZE_BYTES, +}; use crate::types::{ ExecError, ExecRequest, ExecResponse, ExecStats, InvocationType, LogEntry, LogLevel, }; @@ -749,7 +751,11 @@ fn parse_response(value: Dynamic) -> Result<(u16, BTreeMap, Json } } } - Ok((200, BTreeMap::new(), dynamic_to_json(&value))) + // The response body is otherwise an uncapped Rhai→JSON exit — bound it so a + // cheaply-aliased huge value can't OOM the node on materialization. + let body = dynamic_to_json_capped(&value, MAX_JSON_MATERIALIZE_BYTES) + .map_err(|e| ExecError::InvalidResponse(e.to_string()))?; + Ok((200, BTreeMap::new(), body)) } fn parse_structured_response(map: Map) -> Result<(u16, BTreeMap, Json), ExecError> { @@ -771,7 +777,11 @@ fn parse_structured_response(map: Map) -> Result<(u16, BTreeMap, } } - let body = map.get("body").map_or(Json::Null, dynamic_to_json); + let body = match map.get("body") { + Some(b) => dynamic_to_json_capped(b, MAX_JSON_MATERIALIZE_BYTES) + .map_err(|e| ExecError::InvalidResponse(e.to_string()))?, + None => Json::Null, + }; Ok((status_code, headers, body)) } diff --git a/crates/executor-core/src/sdk/bridge.rs b/crates/executor-core/src/sdk/bridge.rs index bdc4b0d..1f48dac 100644 --- a/crates/executor-core/src/sdk/bridge.rs +++ b/crates/executor-core/src/sdk/bridge.rs @@ -85,35 +85,173 @@ pub fn json_to_dynamic(value: Json) -> Dynamic { } } -/// Convert a Rhai `Dynamic` back to a `serde_json::Value`. Custom Rhai -/// types (timestamps, user-registered modules) fall back to their -/// `Display` form so they appear as strings in JSON output rather than -/// failing the response build. +/// Hard ceiling on the byte size of a single Rhai→JSON materialization. +/// +/// The per-element Rhai sandbox caps (`max_string_size` 64 KiB, +/// `max_array_size`/`max_map_size` 10 000) do NOT bound the *materialized* +/// total: Rhai shares strings/arrays by `Rc`, so a 10 000-element array of one +/// aliased 64 KiB string is cheap to build (~10 000 ops) yet dealiases to +/// ~640 MiB of distinct JSON. Without a ceiling a handful of anonymous requests +/// could OOM the node. `dynamic_to_json_capped` charges a running byte budget +/// and bails as soon as it is exceeded, so the transient allocation is bounded +/// to roughly this ceiling regardless of aliasing. This is far above any +/// legitimate single value (the KV/docs value caps are 256 KiB); it is a +/// last-resort anti-OOM rail, not a business limit. +pub const MAX_JSON_MATERIALIZE_BYTES: usize = 16 * 1024 * 1024; + +/// A Rhai value was too large to materialize into JSON within its byte budget. +#[derive(Debug, Clone, Copy)] +pub struct JsonSizeError { + pub limit: usize, +} + +impl std::fmt::Display for JsonSizeError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "value too large to serialize to JSON (exceeds the {}-byte limit)", + self.limit + ) + } +} + +impl std::error::Error for JsonSizeError {} + +impl From for Box { + fn from(e: JsonSizeError) -> Self { + runtime_err(&e.to_string()) + } +} + +/// Convert a Rhai `Dynamic` back to a `serde_json::Value`, **infallibly** and +/// **bounded** by [`MAX_JSON_MATERIALIZE_BYTES`]. A value that would exceed the +/// ceiling collapses to a short marker string rather than OOMing — safe for +/// best-effort paths (logging) where a hard error is undesirable. Paths where +/// silently substituting a marker would be wrong (writes, the HTTP response +/// body, `invoke` args) must use [`dynamic_to_json_capped`] and surface the +/// error instead. +/// +/// Custom Rhai types (timestamps, user-registered modules) fall back to their +/// `Display` form so they appear as strings rather than failing the build. +#[must_use] pub fn dynamic_to_json(value: &Dynamic) -> Json { + dynamic_to_json_capped(value, MAX_JSON_MATERIALIZE_BYTES) + .unwrap_or_else(|_| Json::String("".to_string())) +} + +/// Bounded `Dynamic`→JSON conversion: materializes `value`, charging a running +/// byte budget starting at `max_bytes`, and returns [`JsonSizeError`] the moment +/// the budget is exceeded — so it never builds the full oversized tree. +/// +/// # Errors +/// Returns [`JsonSizeError`] if the materialized JSON would exceed `max_bytes`. +pub fn dynamic_to_json_capped(value: &Dynamic, max_bytes: usize) -> Result { + let mut remaining = max_bytes; + to_json_bounded(value, &mut remaining, max_bytes) +} + +/// Deduct `n` bytes from `remaining`, or fail if the budget is exhausted. +fn charge(remaining: &mut usize, n: usize, limit: usize) -> Result<(), JsonSizeError> { + match remaining.checked_sub(n) { + Some(left) => { + *remaining = left; + Ok(()) + } + None => Err(JsonSizeError { limit }), + } +} + +fn to_json_bounded( + value: &Dynamic, + remaining: &mut usize, + limit: usize, +) -> Result { if value.is_unit() { - return Json::Null; + charge(remaining, 4, limit)?; + return Ok(Json::Null); } if let Ok(b) = value.as_bool() { - return Json::Bool(b); + charge(remaining, 5, limit)?; + return Ok(Json::Bool(b)); } if let Ok(i) = value.as_int() { - return Json::Number(i.into()); + charge(remaining, 8, limit)?; + return Ok(Json::Number(i.into())); } if let Ok(f) = value.as_float() { - return serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number); + charge(remaining, 8, limit)?; + return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number)); } if value.is_string() { - return Json::String(value.clone().into_string().unwrap_or_default()); + let s = value.clone().into_string().unwrap_or_default(); + charge(remaining, s.len() + 2, limit)?; + return Ok(Json::String(s)); } if let Some(arr) = value.clone().try_cast::() { - return Json::Array(arr.iter().map(dynamic_to_json).collect()); + charge(remaining, 2, limit)?; // "[]" + let mut out = Vec::with_capacity(arr.len().min(1024)); + for v in &arr { + charge(remaining, 1, limit)?; // "," + out.push(to_json_bounded(v, remaining, limit)?); + } + return Ok(Json::Array(out)); } if let Some(map) = value.clone().try_cast::() { + charge(remaining, 2, limit)?; // "{}" let mut out = serde_json::Map::new(); for (k, v) in map { - out.insert(k.to_string(), dynamic_to_json(&v)); + charge(remaining, k.len() + 4, limit)?; // "key":, + out.insert(k.to_string(), to_json_bounded(&v, remaining, limit)?); } - return Json::Object(out); + return Ok(Json::Object(out)); + } + let s = value.to_string(); + charge(remaining, s.len() + 2, limit)?; + Ok(Json::String(s)) +} + +#[cfg(test)] +mod tests { + use super::*; + use rhai::{Array, Dynamic}; + + #[test] + fn small_value_round_trips_under_the_cap() { + let d = Dynamic::from("hello"); + assert_eq!( + dynamic_to_json_capped(&d, MAX_JSON_MATERIALIZE_BYTES).unwrap(), + Json::String("hello".into()) + ); + } + + #[test] + fn aliased_large_array_is_rejected_not_materialized() { + // The OOM regression: one 64 KiB string aliased 10 000×. Cheap to build + // (Rc-shared), ~640 MiB if dealiased. The bounded converter must bail on + // the byte budget instead of allocating the full tree. + let big = "x".repeat(64 * 1024); + let mut arr = Array::new(); + for _ in 0..10_000 { + arr.push(Dynamic::from(big.clone())); // ImmutableString: shared, cheap + } + let d = Dynamic::from(arr); + // A 1 MiB budget is far below the ~640 MiB dealiased size → must error. + let err = dynamic_to_json_capped(&d, 1024 * 1024).unwrap_err(); + assert_eq!(err.limit, 1024 * 1024); + } + + #[test] + fn infallible_wrapper_yields_marker_instead_of_oom() { + // ~300 × 64 KiB aliased ≈ 19 MiB dealiased, just over the 16 MiB default + // ceiling — enough to trip the infallible wrapper's marker fallback + // without paying to build a pathologically huge test array. + let big = "x".repeat(64 * 1024); + let mut arr = Array::new(); + for _ in 0..300 { + arr.push(Dynamic::from(big.clone())); + } + // The infallible path must NOT OOM — it collapses to a marker string. + let out = dynamic_to_json(&Dynamic::from(arr)); + assert_eq!(out, Json::String("".into())); } - Json::String(value.to_string()) } diff --git a/crates/executor-core/src/sdk/docs.rs b/crates/executor-core/src/sdk/docs.rs index cc9ea94..05faad7 100644 --- a/crates/executor-core/src/sdk/docs.rs +++ b/crates/executor-core/src/sdk/docs.rs @@ -27,7 +27,9 @@ use picloud_shared::{DocId, DocRow, DocsService, GroupDocsService, SdkCallCx, Se use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; use uuid::Uuid; -use super::bridge::{block_on, dynamic_to_json, json_to_dynamic}; +use super::bridge::{ + block_on, dynamic_to_json_capped, json_to_dynamic, MAX_JSON_MATERIALIZE_BYTES, +}; /// Per-call handle captured by the Rhai SDK. Cheap to clone (two Arcs /// plus an owned string). @@ -117,7 +119,7 @@ fn register_create(engine: &mut RhaiEngine) { "create", |handle: &mut DocsHandle, data: Map| -> Result> { let h = handle.clone(); - let json = dynamic_to_json(&Dynamic::from(data)); + let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?; let id = block_on("docs", async move { h.service.create(&h.cx, &h.collection, json).await })?; @@ -145,7 +147,7 @@ fn register_find(engine: &mut RhaiEngine) { "find", |handle: &mut DocsHandle, filter: Map| -> Result> { let h = handle.clone(); - let json = dynamic_to_json(&Dynamic::from(filter)); + let json = dynamic_to_json_capped(&Dynamic::from(filter), MAX_JSON_MATERIALIZE_BYTES)?; let rows = block_on("docs", async move { h.service.find(&h.cx, &h.collection, json).await })?; @@ -162,7 +164,7 @@ fn register_find_one(engine: &mut RhaiEngine) { "find_one", |handle: &mut DocsHandle, filter: Map| -> Result> { let h = handle.clone(); - let json = dynamic_to_json(&Dynamic::from(filter)); + let json = dynamic_to_json_capped(&Dynamic::from(filter), MAX_JSON_MATERIALIZE_BYTES)?; let row = block_on("docs", async move { h.service.find_one(&h.cx, &h.collection, json).await })?; @@ -177,7 +179,7 @@ fn register_update(engine: &mut RhaiEngine) { |handle: &mut DocsHandle, id: &str, data: Map| -> Result<(), Box> { let h = handle.clone(); let parsed_id = parse_doc_id(id)?; - let json = dynamic_to_json(&Dynamic::from(data)); + let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?; block_on("docs", async move { h.service .update(&h.cx, &h.collection, parsed_id, json) @@ -265,7 +267,7 @@ fn register_group_create(engine: &mut RhaiEngine) { "create", |handle: &mut GroupDocsHandle, data: Map| -> Result> { let h = handle.clone(); - let json = dynamic_to_json(&Dynamic::from(data)); + let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?; let id = block_on("docs", async move { h.service.create(&h.cx, &h.collection, json).await })?; @@ -293,7 +295,7 @@ fn register_group_find(engine: &mut RhaiEngine) { "find", |handle: &mut GroupDocsHandle, filter: Map| -> Result> { let h = handle.clone(); - let json = dynamic_to_json(&Dynamic::from(filter)); + let json = dynamic_to_json_capped(&Dynamic::from(filter), MAX_JSON_MATERIALIZE_BYTES)?; let rows = block_on("docs", async move { h.service.find(&h.cx, &h.collection, json).await })?; @@ -310,7 +312,7 @@ fn register_group_find_one(engine: &mut RhaiEngine) { "find_one", |handle: &mut GroupDocsHandle, filter: Map| -> Result> { let h = handle.clone(); - let json = dynamic_to_json(&Dynamic::from(filter)); + let json = dynamic_to_json_capped(&Dynamic::from(filter), MAX_JSON_MATERIALIZE_BYTES)?; let row = block_on("docs", async move { h.service.find_one(&h.cx, &h.collection, json).await })?; @@ -325,7 +327,7 @@ fn register_group_update(engine: &mut RhaiEngine) { |handle: &mut GroupDocsHandle, id: &str, data: Map| -> Result<(), Box> { let h = handle.clone(); let parsed_id = parse_doc_id(id)?; - let json = dynamic_to_json(&Dynamic::from(data)); + let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?; block_on("docs", async move { h.service .update(&h.cx, &h.collection, parsed_id, json) diff --git a/crates/executor-core/src/sdk/http.rs b/crates/executor-core/src/sdk/http.rs index 520c5a2..a1b95b4 100644 --- a/crates/executor-core/src/sdk/http.rs +++ b/crates/executor-core/src/sdk/http.rs @@ -35,7 +35,9 @@ use std::sync::Arc; use picloud_shared::{HttpRequest, HttpResponse, HttpService, SdkCallCx, Services}; use rhai::{Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; -use super::bridge::{block_on, dynamic_to_json, json_to_dynamic, runtime_err}; +use super::bridge::{ + block_on, dynamic_to_json_capped, json_to_dynamic, runtime_err, MAX_JSON_MATERIALIZE_BYTES, +}; /// Bridge-side defaults (the service clamps server-side too). The /// `MAX_*` ceilings stay `i64` because they're compared against the @@ -236,13 +238,13 @@ fn dispatch_body(body: Dynamic) -> Result> { return Ok((Some(s.into_bytes()), Some("text/plain".to_string()))); } if body.is_map() || body.is_array() { - let json = dynamic_to_json(&body); + let json = dynamic_to_json_capped(&body, MAX_JSON_MATERIALIZE_BYTES)?; let bytes = serde_json::to_vec(&json) .map_err(|e| err(format!("could not encode JSON body: {e}")))?; return Ok((Some(bytes), Some("application/json".to_string()))); } // Scalars (int/float/bool) → JSON-encode for consistency. - let json = dynamic_to_json(&body); + let json = dynamic_to_json_capped(&body, MAX_JSON_MATERIALIZE_BYTES)?; let bytes = serde_json::to_vec(&json).map_err(|e| err(format!("could not encode body: {e}")))?; Ok((Some(bytes), Some("application/json".to_string()))) diff --git a/crates/executor-core/src/sdk/invoke.rs b/crates/executor-core/src/sdk/invoke.rs index aeeb379..c5411b0 100644 --- a/crates/executor-core/src/sdk/invoke.rs +++ b/crates/executor-core/src/sdk/invoke.rs @@ -42,7 +42,7 @@ use tokio::runtime::Handle as TokioHandle; use crate::engine::Engine; use crate::sandbox::Limits; -use crate::sdk::bridge::json_to_dynamic; +use crate::sdk::bridge::{json_to_dynamic, MAX_JSON_MATERIALIZE_BYTES}; use crate::types::{ExecRequest, InvocationType}; pub(super) fn register( @@ -248,8 +248,35 @@ fn parse_target(target: Dynamic) -> Result> { } /// Convert Rhai Dynamic → JSON. Rejects FnPtr at any depth (closures -/// don't survive invoke boundaries). +/// don't survive invoke boundaries) and is bounded by +/// [`MAX_JSON_MATERIALIZE_BYTES`] so a cheaply-aliased huge args value can't OOM +/// the node on materialization (same anti-OOM rail as `dynamic_to_json_capped`). fn args_to_json(value: &Dynamic) -> Result> { + let mut remaining = MAX_JSON_MATERIALIZE_BYTES; + args_to_json_bounded(value, &mut remaining) +} + +fn args_charge(remaining: &mut usize, n: usize) -> Result<(), Box> { + match remaining.checked_sub(n) { + Some(left) => { + *remaining = left; + Ok(()) + } + None => Err(EvalAltResult::ErrorRuntime( + format!( + "invoke: args too large to serialize (exceeds the {MAX_JSON_MATERIALIZE_BYTES}-byte limit)" + ) + .into(), + rhai::Position::NONE, + ) + .into()), + } +} + +fn args_to_json_bounded( + value: &Dynamic, + remaining: &mut usize, +) -> Result> { if value.is::() { return Err(EvalAltResult::ErrorRuntime( "invoke: args must not contain FnPtr / closures".into(), @@ -259,40 +286,52 @@ fn args_to_json(value: &Dynamic) -> Result> { } if value.is_blob() { let blob = value.clone().into_blob().unwrap_or_default(); - return Ok(Json::String(STANDARD.encode(&blob))); + let encoded = STANDARD.encode(&blob); + args_charge(remaining, encoded.len() + 2)?; + return Ok(Json::String(encoded)); } if value.is_unit() { + args_charge(remaining, 4)?; return Ok(Json::Null); } if let Ok(b) = value.as_bool() { + args_charge(remaining, 5)?; return Ok(Json::Bool(b)); } if let Ok(i) = value.as_int() { + args_charge(remaining, 8)?; return Ok(Json::Number(i.into())); } if let Ok(f) = value.as_float() { + args_charge(remaining, 8)?; return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number)); } if value.is_string() { - return Ok(Json::String( - value.clone().into_string().unwrap_or_default(), - )); + let s = value.clone().into_string().unwrap_or_default(); + args_charge(remaining, s.len() + 2)?; + return Ok(Json::String(s)); } if let Some(arr) = value.clone().try_cast::() { - let mut out = Vec::with_capacity(arr.len()); + args_charge(remaining, 2)?; + let mut out = Vec::with_capacity(arr.len().min(1024)); for v in &arr { - out.push(args_to_json(v)?); + args_charge(remaining, 1)?; + out.push(args_to_json_bounded(v, remaining)?); } return Ok(Json::Array(out)); } if let Some(map) = value.clone().try_cast::() { + args_charge(remaining, 2)?; let mut out = serde_json::Map::new(); for (k, v) in map { - out.insert(k.to_string(), args_to_json(&v)?); + args_charge(remaining, k.len() + 4)?; + out.insert(k.to_string(), args_to_json_bounded(&v, remaining)?); } return Ok(Json::Object(out)); } - Ok(Json::String(value.to_string())) + let s = value.to_string(); + args_charge(remaining, s.len() + 2)?; + Ok(Json::String(s)) } /// `Limits` is `Copy`; passed by value into `register` so closures take diff --git a/crates/executor-core/src/sdk/kv.rs b/crates/executor-core/src/sdk/kv.rs index 94fc1b8..911813a 100644 --- a/crates/executor-core/src/sdk/kv.rs +++ b/crates/executor-core/src/sdk/kv.rs @@ -33,7 +33,9 @@ use std::sync::Arc; use picloud_shared::{GroupKvService, InterceptorService, KvService, SdkCallCx, Services}; use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; -use super::bridge::{block_on, dynamic_to_json, json_to_dynamic}; +use super::bridge::{ + block_on, dynamic_to_json_capped, json_to_dynamic, JsonSizeError, MAX_JSON_MATERIALIZE_BYTES, +}; use crate::engine::Engine; use crate::sandbox::Limits; @@ -154,11 +156,14 @@ pub(super) fn register( /// Map a Rhai `expected` argument to the CAS precondition: Rhai unit `()` means /// "expected ABSENT" (insert-if-absent); any other value is the expected current /// value. -fn expected_from_dynamic(expected: &Dynamic) -> Option { +fn expected_from_dynamic(expected: &Dynamic) -> Result, JsonSizeError> { if expected.is_unit() { - None + Ok(None) } else { - Some(dynamic_to_json(expected)) + Ok(Some(dynamic_to_json_capped( + expected, + MAX_JSON_MATERIALIZE_BYTES, + )?)) } } @@ -179,7 +184,7 @@ fn register_set(engine: &mut RhaiEngine) { engine.register_fn( "set", |handle: &mut KvHandle, key: &str, value: Dynamic| -> Result<(), Box> { - let json = dynamic_to_json(&value); + let json = dynamic_to_json_capped(&value, MAX_JSON_MATERIALIZE_BYTES)?; // §9.4 before-op interceptor (allow/deny). A denial errors here and // the write below never runs. super::interceptor::run_before( @@ -213,8 +218,8 @@ fn register_set_if(engine: &mut RhaiEngine) { new: Dynamic| -> Result> { let h = handle.clone(); - let exp = expected_from_dynamic(&expected); - let new = dynamic_to_json(&new); + let exp = expected_from_dynamic(&expected)?; + let new = dynamic_to_json_capped(&new, MAX_JSON_MATERIALIZE_BYTES)?; block_on("kv", async move { h.service.set_if(&h.cx, &h.collection, key, exp, new).await }) @@ -343,7 +348,7 @@ fn register_group_set(engine: &mut RhaiEngine) { engine.register_fn( "set", |handle: &mut GroupKvHandle, key: &str, value: Dynamic| -> Result<(), Box> { - let json = dynamic_to_json(&value); + let json = dynamic_to_json_capped(&value, MAX_JSON_MATERIALIZE_BYTES)?; // §9.4 before-op interceptor also guards shared-collection writes. group_run_before(handle, "set", key, Some(&json))?; let h = handle.clone(); @@ -363,8 +368,8 @@ fn register_group_set_if(engine: &mut RhaiEngine) { new: Dynamic| -> Result> { let h = handle.clone(); - let exp = expected_from_dynamic(&expected); - let new = dynamic_to_json(&new); + let exp = expected_from_dynamic(&expected)?; + let new = dynamic_to_json_capped(&new, MAX_JSON_MATERIALIZE_BYTES)?; block_on("kv", async move { h.service.set_if(&h.cx, &h.collection, key, exp, new).await }) diff --git a/crates/executor-core/src/sdk/mod.rs b/crates/executor-core/src/sdk/mod.rs index 9d1f3be..cf5ff27 100644 --- a/crates/executor-core/src/sdk/mod.rs +++ b/crates/executor-core/src/sdk/mod.rs @@ -30,7 +30,10 @@ pub mod users; pub mod vars; pub mod workflow; -pub use bridge::{dynamic_to_json, json_to_dynamic}; +pub use bridge::{ + dynamic_to_json, dynamic_to_json_capped, json_to_dynamic, JsonSizeError, + MAX_JSON_MATERIALIZE_BYTES, +}; pub use cx::SdkCallCx; use std::sync::Arc; diff --git a/crates/executor-core/src/sdk/pubsub.rs b/crates/executor-core/src/sdk/pubsub.rs index 0b56cdb..400fcd4 100644 --- a/crates/executor-core/src/sdk/pubsub.rs +++ b/crates/executor-core/src/sdk/pubsub.rs @@ -24,7 +24,7 @@ use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; use serde_json::Value as Json; use tokio::runtime::Handle as TokioHandle; -use super::bridge::block_on; +use super::bridge::{block_on, MAX_JSON_MATERIALIZE_BYTES}; pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc) { let svc = services.pubsub.clone(); @@ -35,7 +35,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result<(), Box> { - let json = message_to_json(&message); + let json = message_to_json(&message)?; let svc = svc.clone(); let cx = cx.clone(); block_on("pubsub", async move { @@ -110,7 +110,7 @@ fn shared_publish( subtopic: &str, message: Dynamic, ) -> Result<(), Box> { - let json = message_to_json(&message); + let json = message_to_json(&message)?; let service = h.service.clone(); let cx = h.cx.clone(); let namespace = h.namespace.clone(); @@ -189,38 +189,81 @@ fn mint_token( /// Convert a Rhai `Dynamic` message into JSON, base64-encoding any /// `Blob` (at any nesting depth). Mirrors `bridge::dynamic_to_json` but -/// adds the blob arm the pub/sub wire contract requires. -fn message_to_json(value: &Dynamic) -> Json { +/// adds the blob arm the pub/sub wire contract requires. Bounded by +/// [`MAX_JSON_MATERIALIZE_BYTES`] so a cheaply-aliased huge message can't OOM +/// the node on materialization (the per-message wire cap is enforced later, on +/// the already-materialized value). +fn message_to_json(value: &Dynamic) -> Result> { + let mut remaining = MAX_JSON_MATERIALIZE_BYTES; + message_to_json_bounded(value, &mut remaining) +} + +fn msg_charge(remaining: &mut usize, n: usize) -> Result<(), Box> { + match remaining.checked_sub(n) { + Some(left) => { + *remaining = left; + Ok(()) + } + None => Err(EvalAltResult::ErrorRuntime( + format!("pubsub: message too large to serialize (exceeds the {MAX_JSON_MATERIALIZE_BYTES}-byte limit)").into(), + rhai::Position::NONE, + ) + .into()), + } +} + +fn message_to_json_bounded( + value: &Dynamic, + remaining: &mut usize, +) -> Result> { // Blob must be checked before the generic array path (a Blob is a // `Vec`, distinct from a Rhai `Array`). if value.is_blob() { let blob = value.clone().into_blob().unwrap_or_default(); - return Json::String(STANDARD.encode(&blob)); + let encoded = STANDARD.encode(&blob); + msg_charge(remaining, encoded.len() + 2)?; + return Ok(Json::String(encoded)); } if value.is_unit() { - return Json::Null; + msg_charge(remaining, 4)?; + return Ok(Json::Null); } if let Ok(b) = value.as_bool() { - return Json::Bool(b); + msg_charge(remaining, 5)?; + return Ok(Json::Bool(b)); } if let Ok(i) = value.as_int() { - return Json::Number(i.into()); + msg_charge(remaining, 8)?; + return Ok(Json::Number(i.into())); } if let Ok(f) = value.as_float() { - return serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number); + msg_charge(remaining, 8)?; + return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number)); } if value.is_string() { - return Json::String(value.clone().into_string().unwrap_or_default()); + let s = value.clone().into_string().unwrap_or_default(); + msg_charge(remaining, s.len() + 2)?; + return Ok(Json::String(s)); } if let Some(arr) = value.clone().try_cast::() { - return Json::Array(arr.iter().map(message_to_json).collect()); + msg_charge(remaining, 2)?; + let mut out = Vec::with_capacity(arr.len().min(1024)); + for v in &arr { + msg_charge(remaining, 1)?; + out.push(message_to_json_bounded(v, remaining)?); + } + return Ok(Json::Array(out)); } if let Some(map) = value.clone().try_cast::() { + msg_charge(remaining, 2)?; let mut out = serde_json::Map::new(); for (k, v) in map { - out.insert(k.to_string(), message_to_json(&v)); + msg_charge(remaining, k.len() + 4)?; + out.insert(k.to_string(), message_to_json_bounded(&v, remaining)?); } - return Json::Object(out); + return Ok(Json::Object(out)); } - Json::String(value.to_string()) + let s = value.to_string(); + msg_charge(remaining, s.len() + 2)?; + Ok(Json::String(s)) } diff --git a/crates/executor-core/src/sdk/queue.rs b/crates/executor-core/src/sdk/queue.rs index 245c0c9..f594829 100644 --- a/crates/executor-core/src/sdk/queue.rs +++ b/crates/executor-core/src/sdk/queue.rs @@ -27,6 +27,8 @@ use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; use serde_json::Value as Json; use tokio::runtime::Handle as TokioHandle; +use super::bridge::MAX_JSON_MATERIALIZE_BYTES; + pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc) { let svc = services.queue.clone(); let mut module = Module::new(); @@ -262,8 +264,33 @@ fn parse_opts(opts: &Map) -> Result> { /// Convert a Rhai `Dynamic` into JSON. Mirrors `pubsub::message_to_json` /// — blobs become base64, FnPtr / closures are rejected so a script /// can't accidentally enqueue something that won't survive a trip -/// through Postgres + back through the bridge. +/// through Postgres + back through the bridge. Bounded by +/// [`MAX_JSON_MATERIALIZE_BYTES`] so a cheaply-aliased huge message can't OOM +/// the node on materialization (the per-message payload cap is enforced later, +/// on the already-materialized value). fn message_to_json(value: &Dynamic) -> Result> { + let mut remaining = MAX_JSON_MATERIALIZE_BYTES; + message_to_json_bounded(value, &mut remaining) +} + +fn msg_charge(remaining: &mut usize, n: usize) -> Result<(), Box> { + match remaining.checked_sub(n) { + Some(left) => { + *remaining = left; + Ok(()) + } + None => Err(EvalAltResult::ErrorRuntime( + format!("queue::enqueue: message too large to serialize (exceeds the {MAX_JSON_MATERIALIZE_BYTES}-byte limit)").into(), + rhai::Position::NONE, + ) + .into()), + } +} + +fn message_to_json_bounded( + value: &Dynamic, + remaining: &mut usize, +) -> Result> { if value.is::() { return Err(EvalAltResult::ErrorRuntime( "queue::enqueue: messages must not contain FnPtr / closures".into(), @@ -273,40 +300,52 @@ fn message_to_json(value: &Dynamic) -> Result> { } if value.is_blob() { let blob = value.clone().into_blob().unwrap_or_default(); - return Ok(Json::String(STANDARD.encode(&blob))); + let encoded = STANDARD.encode(&blob); + msg_charge(remaining, encoded.len() + 2)?; + return Ok(Json::String(encoded)); } if value.is_unit() { + msg_charge(remaining, 4)?; return Ok(Json::Null); } if let Ok(b) = value.as_bool() { + msg_charge(remaining, 5)?; return Ok(Json::Bool(b)); } if let Ok(i) = value.as_int() { + msg_charge(remaining, 8)?; return Ok(Json::Number(i.into())); } if let Ok(f) = value.as_float() { + msg_charge(remaining, 8)?; return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number)); } if value.is_string() { - return Ok(Json::String( - value.clone().into_string().unwrap_or_default(), - )); + let s = value.clone().into_string().unwrap_or_default(); + msg_charge(remaining, s.len() + 2)?; + return Ok(Json::String(s)); } if let Some(arr) = value.clone().try_cast::() { - let mut out = Vec::with_capacity(arr.len()); + msg_charge(remaining, 2)?; + let mut out = Vec::with_capacity(arr.len().min(1024)); for v in &arr { - out.push(message_to_json(v)?); + msg_charge(remaining, 1)?; + out.push(message_to_json_bounded(v, remaining)?); } return Ok(Json::Array(out)); } if let Some(map) = value.clone().try_cast::() { + msg_charge(remaining, 2)?; let mut out = serde_json::Map::new(); for (k, v) in map { - out.insert(k.to_string(), message_to_json(&v)?); + msg_charge(remaining, k.len() + 4)?; + out.insert(k.to_string(), message_to_json_bounded(&v, remaining)?); } return Ok(Json::Object(out)); } - Ok(Json::String(value.to_string())) + let s = value.to_string(); + msg_charge(remaining, s.len() + 2)?; + Ok(Json::String(s)) } #[cfg(test)] diff --git a/crates/executor-core/src/sdk/secrets.rs b/crates/executor-core/src/sdk/secrets.rs index d133150..954d6d9 100644 --- a/crates/executor-core/src/sdk/secrets.rs +++ b/crates/executor-core/src/sdk/secrets.rs @@ -21,7 +21,9 @@ use std::sync::Arc; use picloud_shared::{SdkCallCx, SecretsListPage, Services}; use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; -use super::bridge::{block_on, dynamic_to_json, json_to_dynamic, runtime_err}; +use super::bridge::{ + block_on, dynamic_to_json_capped, json_to_dynamic, runtime_err, MAX_JSON_MATERIALIZE_BYTES, +}; pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc) { let svc = services.secrets.clone(); @@ -34,7 +36,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result<(), Box> { - let json = dynamic_to_json(&value); + let json = dynamic_to_json_capped(&value, MAX_JSON_MATERIALIZE_BYTES)?; let svc = svc.clone(); let cx = cx.clone(); block_on("secrets", async move { svc.set(&cx, name, json).await }) diff --git a/crates/executor-core/src/sdk/stdlib/json.rs b/crates/executor-core/src/sdk/stdlib/json.rs index ae557eb..fe37c0b 100644 --- a/crates/executor-core/src/sdk/stdlib/json.rs +++ b/crates/executor-core/src/sdk/stdlib/json.rs @@ -4,7 +4,7 @@ use rhai::{Dynamic, Engine as RhaiEngine, EvalAltResult, Module}; -use crate::sdk::bridge::{dynamic_to_json, json_to_dynamic}; +use crate::sdk::bridge::{dynamic_to_json_capped, json_to_dynamic, MAX_JSON_MATERIALIZE_BYTES}; pub fn register(engine: &mut RhaiEngine) { let mut module = Module::new(); @@ -26,7 +26,7 @@ fn register_stringify(module: &mut Module) { module.set_native_fn( "stringify", |v: Dynamic| -> Result> { - serde_json::to_string(&dynamic_to_json(&v)) + serde_json::to_string(&dynamic_to_json_capped(&v, MAX_JSON_MATERIALIZE_BYTES)?) .map_err(|e| format!("json::stringify: {e}").into()) }, ); @@ -36,7 +36,7 @@ fn register_stringify_pretty(module: &mut Module) { module.set_native_fn( "stringify_pretty", |v: Dynamic| -> Result> { - serde_json::to_string_pretty(&dynamic_to_json(&v)) + serde_json::to_string_pretty(&dynamic_to_json_capped(&v, MAX_JSON_MATERIALIZE_BYTES)?) .map_err(|e| format!("json::stringify_pretty: {e}").into()) }, ); diff --git a/crates/executor-core/src/sdk/workflow.rs b/crates/executor-core/src/sdk/workflow.rs index 1dd1af1..3d0d43c 100644 --- a/crates/executor-core/src/sdk/workflow.rs +++ b/crates/executor-core/src/sdk/workflow.rs @@ -16,7 +16,7 @@ use picloud_shared::{SdkCallCx, Services, WorkflowService}; use rhai::{Dynamic, Engine as RhaiEngine, EvalAltResult, Module}; use tokio::runtime::Handle as TokioHandle; -use crate::sdk::bridge::dynamic_to_json; +use crate::sdk::bridge::{dynamic_to_json_capped, MAX_JSON_MATERIALIZE_BYTES}; pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc) { let svc = services.workflow.clone(); @@ -58,7 +58,7 @@ fn start_blocking( let input_json = if input.is_unit() { serde_json::Value::Null } else { - dynamic_to_json(input) + dynamic_to_json_capped(input, MAX_JSON_MATERIALIZE_BYTES)? }; let handle = TokioHandle::try_current().map_err(|e| -> Box { EvalAltResult::ErrorRuntime(