diff --git a/CLAUDE.md b/CLAUDE.md index b047ab7..646affb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -121,6 +121,7 @@ Environment variables consumed by the `picloud` binary: |---|---|---| | `PICLOUD_BIND` | `0.0.0.0:8080` | HTTP listen address. Port 8080 is owned by another process on this host — override locally. | | `PICLOUD_MAX_CONCURRENT_EXECUTIONS` | `32` | Global concurrency cap on data-plane script executions. Overflow returns HTTP 503 with `Retry-After: 1` immediately (no queue). | +| `PICLOUD_MAX_EMISSIONS_PER_EXECUTION` | `1000` | Per-execution fan-out ceiling: max DURABLE emissions (`invoke_async` + `pubsub::publish_durable` + `queue::enqueue`, incl. shared variants) one execution may make before the SDK call errors. Bounds a one-request outbox-amplification DoS. Re-entrancy aware — a synchronous `invoke()`/interceptor chain shares one budget; a dispatched handler starts fresh. `trigger_depth` bounds chain DEPTH, this bounds fan-out WIDTH. | | `DATABASE_URL` | — | Required. Postgres connection string. | | `PICLOUD_SECRET_KEY` | — | Master encryption key (base64). Required at startup unless dev mode is acknowledged (below). | | `PICLOUD_DEV_MODE` | `false` | `true` enables local-dev conveniences. Without `PICLOUD_SECRET_KEY` it ALSO requires the acknowledgement var below — `PICLOUD_DEV_MODE=true` alone aborts at startup. Also: when no SMTP relay is configured, `email::send` switches from disabled (`NotConfigured`) to an **in-memory dev sink** — sends succeed and the last 100 messages are readable at `GET /api/v1/admin/dev/emails` (instance Owner/Admin only; route exists only in this mode). Never in production. | diff --git a/crates/executor-core/src/engine.rs b/crates/executor-core/src/engine.rs index 24365ff..b26e287 100644 --- a/crates/executor-core/src/engine.rs +++ b/crates/executor-core/src/engine.rs @@ -280,6 +280,13 @@ impl Engine { /// cache hands compiled ASTs in directly; this path skips the /// per-call compile. pub fn execute_ast(&self, ast: &Arc, req: ExecRequest) -> Result { + // Audit #2: bound per-execution durable fan-out. The scope is + // re-entrancy aware — a fresh dispatched execution (a new task on a + // pooled thread) resets the budget, while a synchronous invoke() / + // interceptor re-entry nested in the SAME call stack shares it, so + // fan-out is counted across the whole synchronous chain. Held to the end + // of the call; its Drop re-zeroes the counter on the outermost exit. + let _emit_scope = crate::sdk::emit_budget::EmissionBudgetScope::enter(); let effective_limits = self.limits.with_overrides(&req.sandbox_overrides); let logs: Arc>> = Arc::new(Mutex::new(Vec::new())); let mut engine = build_engine(effective_limits, Some(logs.clone())); diff --git a/crates/executor-core/src/sdk/emit_budget.rs b/crates/executor-core/src/sdk/emit_budget.rs new file mode 100644 index 0000000..c3125dd --- /dev/null +++ b/crates/executor-core/src/sdk/emit_budget.rs @@ -0,0 +1,124 @@ +//! Per-execution fan-out ceiling (audit fix #2). +//! +//! `trigger_depth` bounds how DEEP a trigger/invoke chain can go, but nothing +//! bounded how WIDE a single execution could fan out: one anonymous request +//! running `for i in 0..1_000_000 { invoke_async("w", #{}) }` costs a few Rhai +//! ops per iteration plus one cheap outbox INSERT, so within the op / wall-clock +//! budget it could write ~10^5 durable rows — each dispatched as its own +//! execution — flooding the outbox/queue (a durable amplification DoS). +//! +//! This caps the number of DURABLE emissions (`invoke_async`, `publish_durable`, +//! `enqueue`, and their group-shared variants) a single execution may make. +//! The counter is a thread-local. [`EmissionBudgetScope`] is created around +//! EVERY `execute_ast` and is **re-entrancy aware**: only the OUTERMOST scope +//! on a thread resets the counter. A dispatched handler (queue/trigger/ +//! `invoke_async`/cron/workflow) is a fresh task on a pooled thread, so its +//! scope is outermost → it resets and gets a full budget. A synchronous +//! `invoke()` / interceptor re-entry runs a nested `execute_ast` in the SAME +//! call stack, so its scope is inner → it does NOT reset, and the whole +//! synchronous chain shares one budget (fan-out counted across it). The +//! outermost Drop re-zeroes the counter so a pooled thread never leaks a count +//! into the next task. + +use std::cell::Cell; +use std::sync::LazyLock; + +use rhai::EvalAltResult; + +use crate::sdk::bridge::runtime_err; + +/// Default max durable emissions per execution. Generous for a legit batch +/// producer, far below the ~10^5 a runaway loop can reach within budget. +const DEFAULT_MAX_EMISSIONS: u32 = 1000; + +static MAX_EMISSIONS: LazyLock = LazyLock::new(|| { + std::env::var("PICLOUD_MAX_EMISSIONS_PER_EXECUTION") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&n| n > 0) + .unwrap_or(DEFAULT_MAX_EMISSIONS) +}); + +thread_local! { + static EMISSIONS: Cell = const { Cell::new(0) }; + /// Scope nesting depth on this thread. `0` when no execution is active. + static ACTIVE: Cell = const { Cell::new(0) }; +} + +/// Charge one durable emission against the current execution's budget. Returns +/// an `Err` (aborting the emit) once the ceiling is exceeded. +/// +/// # Errors +/// Returns a Rhai runtime error when the per-execution emission ceiling is hit. +pub(super) fn charge_emission(service: &str) -> Result<(), Box> { + EMISSIONS.with(|c| { + let next = c.get().saturating_add(1); + if next > *MAX_EMISSIONS { + return Err(runtime_err(&format!( + "{service}: per-execution emission limit exceeded (max {} durable \ + invoke_async/publish/enqueue calls per execution)", + *MAX_EMISSIONS + ))); + } + c.set(next); + Ok(()) + }) +} + +/// RAII scope wrapping one `execute_ast`. Re-entrancy aware: the OUTERMOST scope +/// on a thread zeroes the counter on entry and on final exit; nested scopes (a +/// synchronous re-entry in the same call stack) leave it alone, so the chain +/// shares one budget. +pub(crate) struct EmissionBudgetScope; + +impl EmissionBudgetScope { + pub(crate) fn enter() -> Self { + ACTIVE.with(|a| { + if a.get() == 0 { + EMISSIONS.with(|c| c.set(0)); + } + a.set(a.get() + 1); + }); + Self + } +} + +impl Drop for EmissionBudgetScope { + fn drop(&mut self) { + ACTIVE.with(|a| { + let n = a.get().saturating_sub(1); + a.set(n); + if n == 0 { + EMISSIONS.with(|c| c.set(0)); + } + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn outermost_scope_resets_but_nested_shares_budget() { + let outer = EmissionBudgetScope::enter(); + for _ in 0..DEFAULT_MAX_EMISSIONS { + charge_emission("test").expect("under budget"); + } + // A NESTED scope (synchronous re-entry) must NOT reset — the budget is + // already exhausted and stays that way. + { + let _nested = EmissionBudgetScope::enter(); + assert!( + charge_emission("test").is_err(), + "a nested re-entry shares the exhausted budget" + ); + } + // Still exhausted after the nested scope drops (outer still active). + assert!(charge_emission("test").is_err()); + drop(outer); + // A fresh OUTERMOST scope resets. + let _fresh = EmissionBudgetScope::enter(); + charge_emission("test").expect("reset by a fresh outermost scope"); + } +} diff --git a/crates/executor-core/src/sdk/invoke.rs b/crates/executor-core/src/sdk/invoke.rs index c5411b0..7e82bc1 100644 --- a/crates/executor-core/src/sdk/invoke.rs +++ b/crates/executor-core/src/sdk/invoke.rs @@ -85,6 +85,9 @@ pub(super) fn register( module.set_native_fn( "invoke_async", move |target: Dynamic, args: Dynamic| -> Result> { + // Audit #2: count this durable emission against the per-execution + // fan-out ceiling before doing any work. + crate::sdk::emit_budget::charge_emission("invoke_async")?; let target = parse_target(target)?; let args_json = args_to_json(&args)?; let svc = svc.clone(); diff --git a/crates/executor-core/src/sdk/mod.rs b/crates/executor-core/src/sdk/mod.rs index cf5ff27..6d0ce51 100644 --- a/crates/executor-core/src/sdk/mod.rs +++ b/crates/executor-core/src/sdk/mod.rs @@ -16,6 +16,7 @@ pub mod cx; pub mod dead_letters; pub mod docs; pub mod email; +pub mod emit_budget; pub mod files; pub mod http; pub mod interceptor; diff --git a/crates/executor-core/src/sdk/pubsub.rs b/crates/executor-core/src/sdk/pubsub.rs index 400fcd4..69abd47 100644 --- a/crates/executor-core/src/sdk/pubsub.rs +++ b/crates/executor-core/src/sdk/pubsub.rs @@ -35,6 +35,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result<(), Box> { + crate::sdk::emit_budget::charge_emission("pubsub::publish_durable")?; let json = message_to_json(&message)?; let svc = svc.clone(); let cx = cx.clone(); @@ -110,6 +111,7 @@ fn shared_publish( subtopic: &str, message: Dynamic, ) -> Result<(), Box> { + crate::sdk::emit_budget::charge_emission("pubsub::shared_topic")?; let json = message_to_json(&message)?; let service = h.service.clone(); let cx = h.cx.clone(); diff --git a/crates/executor-core/src/sdk/queue.rs b/crates/executor-core/src/sdk/queue.rs index f594829..4b9f1a0 100644 --- a/crates/executor-core/src/sdk/queue.rs +++ b/crates/executor-core/src/sdk/queue.rs @@ -40,6 +40,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result<(), Box> { + crate::sdk::emit_budget::charge_emission("queue::enqueue")?; let json = message_to_json(&message)?; enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default()) }, @@ -54,6 +55,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result<(), Box> { + crate::sdk::emit_budget::charge_emission("queue::enqueue")?; let json = message_to_json(&message)?; let opts = parse_opts(&opts)?; enqueue_blocking(&svc, &cx, name, json, opts) @@ -131,6 +133,7 @@ fn group_enqueue( message: Dynamic, opts: EnqueueOpts, ) -> Result<(), Box> { + crate::sdk::emit_budget::charge_emission("queue::shared_collection")?; let json = message_to_json(&message)?; let handle = TokioHandle::try_current().map_err(|e| -> Box { EvalAltResult::ErrorRuntime(