diff --git a/example.env b/example.env index 10bd1a99..e1bbc0ae 100644 --- a/example.env +++ b/example.env @@ -243,6 +243,41 @@ DATABASE_URL=postgres://postgres:postgres@localhost:5432/oxicloud # Max serialized event payload handed to a plugin, in bytes (default: 262144 = 256 KiB) #OXICLOUD_PLUGIN_MAX_INPUT_BYTES=262144 +# Max plugin invocations running at once across all plugins. Past this, dispatch +# sheds load (drops the event, audit-logged) so plugins can't starve the shared +# blocking pool. (default: 16) +#OXICLOUD_PLUGIN_MAX_CONCURRENT_INVOCATIONS=16 + +# Idle window (seconds) after which a plugin's cached compiled module is dropped +# to reclaim memory; the next event recompiles from the on-disk cache. (default: 300) +#OXICLOUD_PLUGIN_CACHE_IDLE_TTL_SECS=300 + +# Decompressed-byte ceiling enforced while unpacking an install bundle (zip-bomb +# guard; the install route also caps the compressed body at 32 MiB). (default: 67108864 = 64 MiB) +#OXICLOUD_PLUGIN_MAX_BUNDLE_DECOMPRESSED_BYTES=67108864 + +# Directory for per-plugin structured logs, one subdir per plugin id. +# (default: {OXICLOUD_STORAGE_PATH}/.plugin-logs) +#OXICLOUD_PLUGIN_LOG_DIR= + +# Size (bytes) at which a plugin's active events.jsonl rotates into a gzip segment. (default: 5242880 = 5 MiB) +#OXICLOUD_PLUGIN_LOG_MAX_FILE_BYTES=5242880 + +# Coarse ceiling on rotated .gz segments kept per plugin at write time. (default: 10) +#OXICLOUD_PLUGIN_LOG_MAX_SEGMENTS=10 + +# Default age (days) past which rotated log segments are pruned by the sweep; +# overridable per plugin in the admin UI. 0 = purge all rotated segments. (default: 30) +#OXICLOUD_PLUGIN_LOG_RETENTION_DAYS=30 + +# Default aggregate byte cap on kept log segments per plugin (oldest deleted +# first); overridable per plugin. 0 = purge all rotated segments. (default: 268435456 = 256 MiB) +#OXICLOUD_PLUGIN_LOG_TOTAL_MAX_BYTES=268435456 + +# Bounded depth of the log-write queue; a flood past this sheds the oldest batch +# rather than blocking dispatch or growing RAM. (default: 1024) +#OXICLOUD_PLUGIN_LOG_QUEUE_CAPACITY=1024 + # ----------------------------------------------------------------------------- # STORAGE BACKEND # ----------------------------------------------------------------------------- diff --git a/src/application/adapters/plugin_user_lifecycle_hook.rs b/src/application/adapters/plugin_user_lifecycle_hook.rs index 0114539a..8bfe24a4 100644 --- a/src/application/adapters/plugin_user_lifecycle_hook.rs +++ b/src/application/adapters/plugin_user_lifecycle_hook.rs @@ -6,10 +6,11 @@ //! it is always compiled and the Extism dependency stays in the infrastructure //! layer. //! -//! Privacy note: the `user.login` payload includes the user's email — PII handed -//! to untrusted plugins with no permission gate in M0. This is acceptable only -//! because plugins are admin-installed today; when the permissions system lands, -//! sensitive payload fields should be gated behind a granted permission. +//! Privacy note: the `user.login` payload is deliberately minimal — an opaque +//! `user_id` plus two non-identifying booleans (`first_login`, `is_external`). +//! It carries no email or username, so no PII reaches untrusted plugins in M0. +//! When the permissions system lands, richer fields (email, username) can be +//! added back behind a granted permission. use std::sync::Arc; @@ -46,8 +47,6 @@ impl UserLifecycleHook for PluginUserLifecycleHook { invocation_id: Uuid::new_v4().to_string(), payload: serde_json::json!({ "user_id": user.id().to_string(), - "username": user.username(), - "email": user.email(), "first_login": user.last_login_at().is_none(), "is_external": user.is_external(), }), @@ -126,10 +125,15 @@ mod tests { let ev = &events[0]; assert_eq!(ev.name, EVENT_USER_LOGIN); assert_eq!(ev.user_id.as_deref(), Some(user.id().to_string().as_str())); - assert_eq!(ev.payload["email"], "alice@example.com"); - assert_eq!(ev.payload["username"], "alice"); + assert_eq!(ev.payload["user_id"], user.id().to_string()); assert_eq!(ev.payload["first_login"], true); // last_login_at is None assert_eq!(ev.payload["is_external"], false); + // Minimal payload: no PII fields. + assert!(ev.payload.get("email").is_none(), "must not leak email"); + assert!( + ev.payload.get("username").is_none(), + "must not leak username" + ); } #[tokio::test] diff --git a/src/application/dtos/plugin_dto.rs b/src/application/dtos/plugin_dto.rs index 76d48b8f..0f12a240 100644 --- a/src/application/dtos/plugin_dto.rs +++ b/src/application/dtos/plugin_dto.rs @@ -104,11 +104,15 @@ pub struct PluginLogQueryDto { } /// Per-plugin retention policy (request + response body). +/// +/// Both limits are accepted as-is, including `0`, which means "purge all rotated +/// segments on the next sweep" (the active log file is never touched). This is +/// intentional — an operator can deliberately keep nothing. #[derive(Debug, Clone, Copy, Serialize, Deserialize, ToSchema)] pub struct PluginRetentionDto { - /// Delete rotated segments older than this many days. + /// Delete rotated segments older than this many days. `0` = keep none. pub retention_days: u32, - /// Aggregate byte ceiling on kept segments for the plugin. + /// Aggregate byte ceiling on kept segments for the plugin. `0` = keep none. pub max_bytes: u64, } diff --git a/src/application/ports/plugin_ports.rs b/src/application/ports/plugin_ports.rs index 4108cf69..c67df670 100644 --- a/src/application/ports/plugin_ports.rs +++ b/src/application/ports/plugin_ports.rs @@ -195,7 +195,7 @@ pub enum PluginMgmtError { /// The bundle failed manifest or runtime validation. Carries the stable /// reason key from `ManifestError::reason()` / `InvokeOutcome::reason()`, /// plus a few install-only keys (`bad_id`, `bad_entrypoint`, `bad_zip`, - /// `no_manifest_in_zip`, `entrypoint_not_in_zip`). + /// `no_manifest_in_zip`, `entrypoint_not_in_zip`, `too_large`). Rejected(&'static str), /// A filesystem error while writing or removing the plugin. Io(String), diff --git a/src/common/config.rs b/src/common/config.rs index ea7015ba..a6f42c28 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -980,6 +980,23 @@ pub struct PluginConfig { /// sweep deletes oldest-first past this. Overridable per plugin. Default: /// 256 MiB. Env: `OXICLOUD_PLUGIN_LOG_TOTAL_MAX_BYTES`. pub log_total_max_bytes: u64, + /// Max plugin invocations running concurrently across all plugins. Dispatch + /// sheds load (drops the event, audit-logged) past this rather than + /// unbounded `spawn_blocking`, so plugins can't starve the shared blocking + /// pool. Default: 16. Env: `OXICLOUD_PLUGIN_MAX_CONCURRENT_INVOCATIONS`. + pub max_concurrent_invocations: usize, + /// Bounded depth of the log-store command channel. A flood past this drops + /// the oldest-arriving log batch (never blocks dispatch). Default: 1024. + /// Env: `OXICLOUD_PLUGIN_LOG_QUEUE_CAPACITY`. + pub log_queue_capacity: usize, + /// Idle window after which a plugin's cached compiled module is dropped to + /// reclaim memory; the next event recompiles from wasmtime's on-disk cache. + /// Default: 300 (5 min). Env: `OXICLOUD_PLUGIN_CACHE_IDLE_TTL_SECS`. + pub cache_idle_ttl_secs: u64, + /// Aggregate decompressed-byte ceiling enforced while unpacking an install + /// bundle (zip-bomb guard; the install route also caps the compressed body). + /// Default: 64 MiB. Env: `OXICLOUD_PLUGIN_MAX_BUNDLE_DECOMPRESSED_BYTES`. + pub max_bundle_decompressed_bytes: u64, } impl Default for PluginConfig { @@ -995,6 +1012,10 @@ impl Default for PluginConfig { log_max_segments: 10, log_retention_days: 30, log_total_max_bytes: 256 * 1024 * 1024, + max_concurrent_invocations: 16, + log_queue_capacity: 1024, + cache_idle_ttl_secs: 300, + max_bundle_decompressed_bytes: 64 * 1024 * 1024, } } } @@ -1435,6 +1456,28 @@ impl AppConfig { { config.plugins.log_total_max_bytes = val; } + if let Ok(v) = + env::var("OXICLOUD_PLUGIN_MAX_CONCURRENT_INVOCATIONS").map(|v| v.parse::()) + && let Ok(val) = v + { + config.plugins.max_concurrent_invocations = val; + } + if let Ok(v) = env::var("OXICLOUD_PLUGIN_LOG_QUEUE_CAPACITY").map(|v| v.parse::()) + && let Ok(val) = v + { + config.plugins.log_queue_capacity = val; + } + if let Ok(v) = env::var("OXICLOUD_PLUGIN_CACHE_IDLE_TTL_SECS").map(|v| v.parse::()) + && let Ok(val) = v + { + config.plugins.cache_idle_ttl_secs = val; + } + if let Ok(v) = + env::var("OXICLOUD_PLUGIN_MAX_BUNDLE_DECOMPRESSED_BYTES").map(|v| v.parse::()) + && let Ok(val) = v + { + config.plugins.max_bundle_decompressed_bytes = val; + } if let Ok(v) = env::var("OXICLOUD_EXPOSE_SYSTEM_USERS").map(|v| v.parse::()) && let Ok(val) = v diff --git a/src/common/di.rs b/src/common/di.rs index e011f2e7..62a3e133 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -650,6 +650,20 @@ impl AppServiceFactory { ) .start(); + // Periodic idle-eviction of cached compiled modules: frees the + // memory of plugins not invoked within the configured TTL; the + // next event recompiles transparently. Cheap, so it ticks often. + { + let evictor = manager.clone(); + tokio::spawn(async move { + let mut tick = tokio::time::interval(std::time::Duration::from_secs(60)); + loop { + tick.tick().await; + evictor.evict_idle_compiled(); + } + }); + } + return (Some(dispatch), Some(management)); } } diff --git a/src/infrastructure/services/plugins/log_store.rs b/src/infrastructure/services/plugins/log_store.rs index ab216cdb..d81051c6 100644 --- a/src/infrastructure/services/plugins/log_store.rs +++ b/src/infrastructure/services/plugins/log_store.rs @@ -41,8 +41,6 @@ use crate::application::ports::plugin_ports::{ const ACTIVE_FILE: &str = "events.jsonl"; /// Marker file holding a plugin's retention override. const RETENTION_FILE: &str = "retention.json"; -/// Bounded command-channel depth — backpressure under flood, not unbounded RAM. -const CHANNEL_CAPACITY: usize = 1024; /// Live broadcast buffer; a slow tailer past this gets `Lagged` (never blocks). const LIVE_CAPACITY: usize = 256; @@ -88,14 +86,17 @@ pub struct PluginLogStore { impl PluginLogStore { /// Spawn the actor thread and return a handle. `default_retention` is applied - /// to any plugin lacking an explicit `retention.json`. + /// to any plugin lacking an explicit `retention.json`. `queue_capacity` + /// bounds the command channel — a flood past it sheds the oldest-arriving + /// batch rather than growing RAM or blocking dispatch. pub fn new( root: PathBuf, max_file_bytes: u64, max_segments: u32, default_retention: RetentionSettings, + queue_capacity: usize, ) -> Self { - let (tx, rx) = mpsc::channel(CHANNEL_CAPACITY); + let (tx, rx) = mpsc::channel(queue_capacity.max(1)); let (live, _) = broadcast::channel(LIVE_CAPACITY); let actor = Actor { root, @@ -115,10 +116,10 @@ impl PluginLogStore { } /// Enqueue a batch (plugin-emitted lines + the host outcome) for one - /// invocation. Called from the dispatch `spawn_blocking` closure, so - /// `blocking_send` is correct: it applies backpressure to that off-runtime - /// thread and preserves order. Send failures (actor gone) are swallowed — - /// logging must never break dispatch. + /// invocation. Called from the dispatch `spawn_blocking` closure. Uses a + /// non-blocking `try_send`: under flood it sheds the batch (logged) rather + /// than blocking the blocking-pool thread or growing RAM unboundedly. A full + /// queue or a gone actor is swallowed — logging must never break dispatch. pub fn append( &self, plugin_id: &str, @@ -148,7 +149,7 @@ impl PluginLogStore { msg, }); - if let Err(e) = self.tx.blocking_send(LogCommand::Append { + if let Err(e) = self.tx.try_send(LogCommand::Append { plugin_id: plugin_id.to_string(), entries, }) { @@ -156,7 +157,7 @@ impl PluginLogStore { target: "oxicloud::plugins", plugin_id = %plugin_id, error = %e, - "dropping plugin log batch: log actor unavailable" + "dropping plugin log batch: queue full or log actor unavailable" ); } } diff --git a/src/infrastructure/services/plugins/manager.rs b/src/infrastructure/services/plugins/manager.rs index 60057cdd..f695487e 100644 --- a/src/infrastructure/services/plugins/manager.rs +++ b/src/infrastructure/services/plugins/manager.rs @@ -17,8 +17,10 @@ use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::{Arc, RwLock}; +use std::time::Duration; use async_trait::async_trait; +use tokio::sync::Semaphore; use super::log_store::PluginLogStore; use super::manifest; @@ -74,6 +76,9 @@ pub struct ExtismPluginManager { plugins: RwLock>, /// Per-plugin structured log storage (shared with the maintenance task). log_store: Arc, + /// Caps concurrent plugin invocations across all plugins so dispatch can + /// shed load instead of flooding the shared blocking pool. + invocation_sem: Arc, } impl ExtismPluginManager { @@ -96,7 +101,9 @@ impl ExtismPluginManager { retention_days: config.log_retention_days, max_bytes: config.log_total_max_bytes, }, + config.log_queue_capacity, )); + let invocation_sem = Arc::new(Semaphore::new(config.max_concurrent_invocations.max(1))); let mut plugins = Vec::new(); let mut rejected = 0usize; @@ -115,6 +122,7 @@ impl ExtismPluginManager { root_dir: dir.to_path_buf(), plugins: RwLock::new(plugins), log_store, + invocation_sem, }; } }; @@ -164,6 +172,7 @@ impl ExtismPluginManager { root_dir: dir.to_path_buf(), plugins: RwLock::new(plugins), log_store, + invocation_sem, } } @@ -183,6 +192,13 @@ impl ExtismPluginManager { std::fs::read_to_string(&manifest_path).map_err(|_| "manifest_unreadable")?; let manifest = manifest::parse_and_validate(&toml_str).map_err(|e| e.reason())?; + // The entrypoint becomes a path joined onto the plugin dir; reject a + // traversal-unsafe value on disk too (mirrors the `install` check), so a + // hand-placed manifest can't read a `.wasm` outside its own directory. + if !is_safe_component(&manifest.plugin.entrypoint) { + return Err("bad_entrypoint"); + } + let wasm_path = dir.join(&manifest.plugin.entrypoint); let wasm_bytes = std::fs::read(&wasm_path).map_err(|_| "wasm_unreadable")?; @@ -229,6 +245,26 @@ impl ExtismPluginManager { self.read_plugins().len() } + /// Drop the cached compiled module of every plugin idle past the configured + /// TTL, reclaiming memory. Driven by a periodic timer in DI; the next event + /// to a freed plugin recompiles transparently. + pub fn evict_idle_compiled(&self) { + let ttl = Duration::from_secs(self.config.cache_idle_ttl_secs); + let mut evicted = 0usize; + for plugin in self.read_plugins().iter() { + if plugin.runtime.evict_if_idle(ttl) { + evicted += 1; + } + } + if evicted > 0 { + tracing::debug!( + target: "oxicloud::plugins", + evicted, + "evicted idle compiled plugin modules" + ); + } + } + fn read_plugins(&self) -> std::sync::RwLockReadGuard<'_, Vec> { self.plugins.read().unwrap_or_else(|e| e.into_inner()) } @@ -279,6 +315,26 @@ impl PluginDispatchPort for ExtismPluginManager { } }; + // Load shedding: cap concurrent invocations so a flood of events (or + // slow plugins) can't exhaust the shared blocking pool. Past the cap + // the event is dropped — plugins are observe-only, so shedding is + // safe; we just record it. + let permit = match self.invocation_sem.clone().try_acquire_owned() { + Ok(p) => p, + Err(_) => { + tracing::warn!( + target: "audit", + event = "plugin.dispatch_shed", + reason = "at_capacity", + plugin_id = %plugin.id, + invocation_id = %event.invocation_id, + plugin_event = %event.name, + "👮🏻‍♂️ plugin event dropped: invocation limit reached" + ); + continue; + } + }; + let runtime = plugin.runtime.clone(); let config = self.config.clone(); let plugin_id = plugin.id.clone(); @@ -289,6 +345,8 @@ impl PluginDispatchPort for ExtismPluginManager { // Run the synchronous wasm call off the async workers. Fire-and-forget: // the upload already succeeded; plugins are post-hoc observers. tokio::task::spawn_blocking(move || { + // Hold the permit for the lifetime of the invocation. + let _permit = permit; let result = runtime.invoke(&config, &export, &invocation_id, &input_json); // Persist every invocation (the plugin's own log lines plus the // host outcome) to the plugin's structured log. Ordered, async, @@ -415,6 +473,11 @@ impl PluginManagementPort for ExtismPluginManager { fn install_bundle(&self, zip: Vec) -> Result { use std::io::{Cursor, Read}; + // Aggregate decompressed ceiling, enforced as each entry is unpacked so + // a zip bomb can't blow up memory before validation (the route also caps + // the compressed body). We only ever extract two named entries. + let max_decompressed: u64 = self.config.max_bundle_decompressed_bytes; + let mut archive = zip::ZipArchive::new(Cursor::new(zip)) .map_err(|_| PluginMgmtError::Rejected("bad_zip"))?; @@ -427,11 +490,18 @@ impl PluginManagementPort for ExtismPluginManager { .ok_or(PluginMgmtError::Rejected("no_manifest_in_zip"))?; let mut manifest_toml = String::new(); - archive - .by_name(&manifest_name) - .map_err(|_| PluginMgmtError::Rejected("no_manifest_in_zip"))? - .read_to_string(&mut manifest_toml) - .map_err(|_| PluginMgmtError::Rejected("bad_zip"))?; + { + let entry = archive + .by_name(&manifest_name) + .map_err(|_| PluginMgmtError::Rejected("no_manifest_in_zip"))?; + entry + .take(max_decompressed + 1) + .read_to_string(&mut manifest_toml) + .map_err(|_| PluginMgmtError::Rejected("bad_zip"))?; + } + if manifest_toml.len() as u64 > max_decompressed { + return Err(PluginMgmtError::Rejected("too_large")); + } // Parse just to learn the entrypoint name; `install` does the full // validation (and rejects a traversal-unsafe entrypoint). @@ -445,12 +515,21 @@ impl PluginManagementPort for ExtismPluginManager { }; let wasm_name = format!("{prefix}{}", manifest.plugin.entrypoint); + // Budget the wasm against what the manifest already consumed. + let remaining = max_decompressed - manifest_toml.len() as u64; let mut wasm = Vec::new(); - archive - .by_name(&wasm_name) - .map_err(|_| PluginMgmtError::Rejected("entrypoint_not_in_zip"))? - .read_to_end(&mut wasm) - .map_err(|_| PluginMgmtError::Rejected("bad_zip"))?; + { + let entry = archive + .by_name(&wasm_name) + .map_err(|_| PluginMgmtError::Rejected("entrypoint_not_in_zip"))?; + entry + .take(remaining + 1) + .read_to_end(&mut wasm) + .map_err(|_| PluginMgmtError::Rejected("bad_zip"))?; + } + if wasm.len() as u64 > remaining { + return Err(PluginMgmtError::Rejected("too_large")); + } self.install(&manifest_toml, wasm) } diff --git a/src/infrastructure/services/plugins/manager_test.rs b/src/infrastructure/services/plugins/manager_test.rs index f55e53aa..e54dc932 100644 --- a/src/infrastructure/services/plugins/manager_test.rs +++ b/src/infrastructure/services/plugins/manager_test.rs @@ -147,6 +147,25 @@ fn install_bundle_missing_entrypoint_is_rejected() { assert_eq!(mgr.loaded_count(), 0); } +#[test] +fn install_bundle_oversized_is_rejected() { + let tmp = tempfile::tempdir().unwrap(); + // Tiny decompressed ceiling so the ~130 KiB wasm fixture trips it cheaply. + let mut config = cfg(); + config.max_bundle_decompressed_bytes = 1024; + let mgr = ExtismPluginManager::load_from_dir(config, tmp.path()); + + let zip = make_zip(&[ + ("plugin.toml", hello_manifest().as_bytes()), + ("hello.wasm", &fixture("hello.wasm")), + ]); + let err = mgr + .install_bundle(zip) + .expect_err("a bundle over the decompressed ceiling must be rejected"); + assert_eq!(err.reason(), "too_large"); + assert_eq!(mgr.loaded_count(), 0); +} + #[test] fn install_bundle_with_garbage_is_rejected() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/infrastructure/services/plugins/manifest.rs b/src/infrastructure/services/plugins/manifest.rs index bb35f601..780c872a 100644 --- a/src/infrastructure/services/plugins/manifest.rs +++ b/src/infrastructure/services/plugins/manifest.rs @@ -40,7 +40,8 @@ pub struct PluginSection { #[derive(Debug, Clone, serde::Deserialize)] #[serde(deny_unknown_fields)] pub struct EventsSection { - /// Events this plugin wants. M0 accepts only `"file.uploaded"`. + /// Events this plugin wants. Each must be one of `KNOWN_EVENTS` + /// (`"file.uploaded"`, `"user.login"`); an unknown name rejects the plugin. pub subscribe: Vec, } diff --git a/src/infrastructure/services/plugins/runtime.rs b/src/infrastructure/services/plugins/runtime.rs index 2ce892c0..b170acc1 100644 --- a/src/infrastructure/services/plugins/runtime.rs +++ b/src/infrastructure/services/plugins/runtime.rs @@ -1,46 +1,76 @@ -//! The Extism runtime wrapper — one sandboxed, per-invocation WASM instance. +//! The Extism runtime wrapper — a cached compiled module, instantiated fresh per +//! invocation. //! //! Isolation is the point: no WASI, no filesystem, no network, a memory cap, and //! a wall-clock timeout. The only authority a plugin has is the host `log` //! function. Every boundary crossing is wrapped so a trap/timeout/OOM/malformed //! output is captured as an [`InvokeOutcome`] and never propagates to the caller. +//! +//! **Compilation is amortized.** A plugin's WASM is compiled once into an +//! [`extism::CompiledPlugin`] and cached; every invocation builds a *fresh* +//! [`extism::Plugin`] instance from it (a new Store/memory → no cross-user +//! state), but pays no recompilation. Per-invocation log attribution rides +//! `call_with_host_context` rather than a baked `UserData`, so the same compiled +//! module serves concurrent invocations without sharing the log buffer. An idle +//! plugin's compiled module is dropped by [`PluginRuntime::evict_if_idle`] to +//! reclaim memory; the next event recompiles (cheaply, from wasmtime's on-disk +//! compilation cache). -use std::time::Duration; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::{Duration, Instant}; -use extism::{Manifest as ExtismManifest, PTR, PluginBuilder, UserData, Wasm}; +use extism::{ + CompiledPlugin, CurrentPlugin, Manifest as ExtismManifest, PTR, PluginBuilder, UserData, Val, + Wasm, +}; use crate::application::ports::plugin_ports::{HOST_NAMESPACE, OXICLOUD_PLUGIN_ABI, PluginOutput}; use crate::common::config::PluginConfig; -/// Per-invocation host state: the plugin's identity (for log attribution) plus -/// the buffer the `log` host function appends to. Shared with the running -/// instance via [`UserData`]; read back after the call via [`drain`]. -#[derive(Default)] -pub struct LogContext { - pub plugin_id: String, - pub invocation_id: String, - pub lines: Vec<(String, String)>, +/// Per-invocation host context, handed to one `handle` call via +/// `call_with_host_context` and read back by the `log` host function. Each +/// invocation gets its own, so a reused compiled module never mixes two +/// invocations' log lines. `lines` is an `Arc` the caller retains a clone of, to +/// read what the plugin emitted after the call returns. +struct LogSink { + plugin_id: String, + invocation_id: String, + lines: Arc>>, } -// The entire authority surface: log(level, message) -> (). Observe-only — it -// reads nothing and mutates no host state. Unknown levels clamp to "info". -extism::host_fn!(oxi_log(user_data: LogContext; level: String, message: String) { +/// The entire authority surface: log(level, message) -> (). Observe-only — it +/// reads nothing and mutates no host state beyond the per-call sink. Unknown +/// levels clamp to "info". Written without the `host_fn!` macro so it can read +/// the per-invocation [`LogSink`] from the host context. +fn oxi_log( + plugin: &mut CurrentPlugin, + inputs: &[Val], + _outputs: &mut [Val], + _user_data: UserData<()>, +) -> Result<(), extism::Error> { + let level: String = plugin.memory_get_val(&inputs[0])?; + let message: String = plugin.memory_get_val(&inputs[1])?; let level = match level.as_str() { "debug" | "info" | "warn" | "error" => level, _ => "info".to_string(), }; - let ud = user_data.get()?; - let mut ctx = ud.lock().unwrap(); + let ctx = plugin.host_context::()?; + // The message is a structured field, never interpolated into the format + // string — a plugin can't inject newlines into the operational log stream. tracing::info!( target: "oxicloud::plugins", plugin_id = %ctx.plugin_id, invocation_id = %ctx.invocation_id, plugin_level = %level, - "plugin log: {message}" + plugin_message = %message, + "plugin log" ); - ctx.lines.push((level, message)); + ctx.lines + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push((level, message)); Ok(()) -}); +} /// The result of one boundary crossing. Only `Ok` is a success; every other /// variant is a contained failure the host audit-logs and moves past. @@ -114,11 +144,20 @@ pub struct InvokeResult { pub logs: Vec<(String, String)>, } -/// A loaded-but-not-instantiated plugin: the wasm bytes plus identity. A fresh -/// instance is built for every invocation (no reuse → no cross-user state). +/// A loaded plugin: the wasm bytes plus a lazily-built, idle-evictable compiled +/// module. A fresh *instance* is built for every invocation (no reuse → no +/// cross-user state); only the *compilation* is shared. pub struct PluginRuntime { plugin_id: String, wasm_bytes: Vec, + /// The cached compiled module, `None` until first use or after idle + /// eviction. Guarded by an `RwLock`: invocations take the read lock to + /// instantiate concurrently; (re)compilation and eviction take the write + /// lock. + compiled: RwLock>, + /// Last time an instance was built, for idle eviction. Separate lock so it + /// can be stamped while only holding `compiled` for read. + last_used: Mutex, } impl PluginRuntime { @@ -126,19 +165,15 @@ impl PluginRuntime { Self { plugin_id: plugin_id.into(), wasm_bytes, + compiled: RwLock::new(None), + last_used: Mutex::new(Instant::now()), } } - pub fn plugin_id(&self) -> &str { - &self.plugin_id - } - - /// Build a fresh, fully locked-down instance for one invocation. - fn build( - &self, - cfg: &PluginConfig, - logs: UserData, - ) -> Result { + /// Compile the WASM into a reusable [`CompiledPlugin`], wiring the sandbox + /// limits and the sole host import. wasmtime's on-disk cache (extism's + /// default) makes a repeat compile after eviction cheap. + fn compile(&self, cfg: &PluginConfig) -> Result { let manifest = ExtismManifest::new([Wasm::data(self.wasm_bytes.clone())]) .with_memory_max(cfg.max_memory_pages) // pages × 64 KiB .with_timeout(Duration::from_millis(cfg.invocation_timeout_ms)) @@ -146,19 +181,68 @@ impl PluginRuntime { // No allowed_paths -> no filesystem. with_wasi(false) -> no ambient authority. PluginBuilder::new(manifest) .with_wasi(false) - .with_function_in_namespace(HOST_NAMESPACE, "log", [PTR, PTR], [], logs, oxi_log) - .build() + .with_function_in_namespace( + HOST_NAMESPACE, + "log", + [PTR, PTR], + [], + UserData::new(()), + oxi_log, + ) + .compile() } - /// Probe a throwaway instance at load time: check `abi_version`, then verify - /// every `required_export` (the `on_` symbol for each subscribed - /// event) actually exists in the module. Rejects lying, unloadable, or + /// Build a fresh instance from the (cached, lazily-compiled) module. Stamps + /// `last_used` so the idle sweep leaves an actively-used plugin alone. + fn instantiate(&self, cfg: &PluginConfig) -> Result { + // Fast path: already compiled. + { + let guard = self.compiled.read().unwrap_or_else(|e| e.into_inner()); + if let Some(compiled) = guard.as_ref() { + *self.last_used.lock().unwrap_or_else(|e| e.into_inner()) = Instant::now(); + return extism::Plugin::new_from_compiled(compiled) + .map_err(|e| InvokeOutcome::LoadError(e.to_string())); + } + } + // Slow path: compile under the write lock (double-checked). + let mut guard = self.compiled.write().unwrap_or_else(|e| e.into_inner()); + if guard.is_none() { + match self.compile(cfg) { + Ok(c) => *guard = Some(c), + Err(e) => return Err(InvokeOutcome::LoadError(e.to_string())), + } + } + let compiled = guard.as_ref().expect("compiled present after compile"); + *self.last_used.lock().unwrap_or_else(|e| e.into_inner()) = Instant::now(); + extism::Plugin::new_from_compiled(compiled) + .map_err(|e| InvokeOutcome::LoadError(e.to_string())) + } + + /// Drop the cached compiled module if it hasn't been used within `ttl`, + /// reclaiming its memory. Returns whether anything was evicted. The next + /// invocation recompiles transparently. + pub fn evict_if_idle(&self, ttl: Duration) -> bool { + let idle = self + .last_used + .lock() + .unwrap_or_else(|e| e.into_inner()) + .elapsed() + >= ttl; + if !idle { + return false; + } + let mut guard = self.compiled.write().unwrap_or_else(|e| e.into_inner()); + guard.take().is_some() + } + + /// Probe loadability: compile (caching the module), check `abi_version`, + /// then verify every `required_export` (the `on_` symbol for each + /// subscribed event) exists. Rejects lying, unloadable, or /// incompletely-implemented plugins before they are ever registered. pub fn check_loadable(&self, cfg: &PluginConfig, required_exports: &[String]) -> InvokeOutcome { - let logs = UserData::new(LogContext::default()); - let mut plugin = match self.build(cfg, logs) { + let mut plugin = match self.instantiate(cfg) { Ok(p) => p, - Err(e) => return InvokeOutcome::LoadError(e.to_string()), + Err(o) => return o, }; match plugin.call::<(), u32>("abi_version", ()) { Ok(v) if v == OXICLOUD_PLUGIN_ABI => {} @@ -192,41 +276,46 @@ impl PluginRuntime { }; } - let logs = UserData::new(LogContext { - plugin_id: self.plugin_id.clone(), - invocation_id: invocation_id.to_string(), - lines: Vec::new(), - }); + let lines = Arc::new(Mutex::new(Vec::new())); + let drain = || lines.lock().unwrap_or_else(|e| e.into_inner()).clone(); - let mut plugin = match self.build(cfg, logs.clone()) { + let mut plugin = match self.instantiate(cfg) { Ok(p) => p, - Err(e) => { + Err(outcome) => { return InvokeResult { - outcome: InvokeOutcome::LoadError(e.to_string()), - logs: drain(&logs), + outcome, + logs: drain(), }; } }; - // Version negotiation at the door. + // Version negotiation at the door (cheap; no recompile). match plugin.call::<(), u32>("abi_version", ()) { Ok(v) if v == OXICLOUD_PLUGIN_ABI => {} Ok(v) => { return InvokeResult { outcome: InvokeOutcome::AbiMismatch { got: v }, - logs: drain(&logs), + logs: drain(), }; } Err(e) => { return InvokeResult { outcome: classify_call_error(e), - logs: drain(&logs), + logs: drain(), }; } } + let sink = LogSink { + plugin_id: self.plugin_id.clone(), + invocation_id: invocation_id.to_string(), + lines: lines.clone(), + }; + // The actual call. Traps, timeouts, and OOM all surface here as Err. - let outcome = match plugin.call::<&str, String>(export, input_json) { + let outcome = match plugin + .call_with_host_context::<&str, String, LogSink>(export, input_json, sink) + { Ok(out) => match serde_json::from_str::(&out) { Ok(parsed) if parsed.ok => InvokeOutcome::Ok, Ok(parsed) => { @@ -239,9 +328,10 @@ impl PluginRuntime { InvokeResult { outcome, - logs: drain(&logs), + logs: drain(), } - // `plugin` dropped here -> sandbox memory reclaimed. + // `plugin` (instance) dropped here -> sandbox memory reclaimed. The + // compiled module stays cached for the next invocation. } } @@ -255,10 +345,3 @@ fn classify_call_error(e: extism::Error) -> InvokeOutcome { InvokeOutcome::Trap(msg) } } - -fn drain(logs: &UserData) -> Vec<(String, String)> { - logs.get() - .ok() - .map(|m| m.lock().unwrap().lines.clone()) - .unwrap_or_default() -} diff --git a/src/infrastructure/services/plugins/runtime_test.rs b/src/infrastructure/services/plugins/runtime_test.rs index d27bcf7a..de7e6afa 100644 --- a/src/infrastructure/services/plugins/runtime_test.rs +++ b/src/infrastructure/services/plugins/runtime_test.rs @@ -172,6 +172,38 @@ fn enforces_timeout() { ); } +#[test] +fn idle_eviction_drops_and_recompiles() { + let rt = PluginRuntime::new("com.example.hello", fixture("hello.wasm")); + // First invoke compiles + caches the module. + let r1 = rt.invoke(&cfg(), "on_file_uploaded", "inv1", &file_uploaded_input()); + assert!(r1.outcome.is_ok(), "first invoke: {:?}", r1.outcome); + + // Idle past a zero TTL -> the cached module is dropped. + assert!( + rt.evict_if_idle(Duration::ZERO), + "a just-idle module should be evicted" + ); + // Nothing left to evict the second time. + assert!( + !rt.evict_if_idle(Duration::ZERO), + "second eviction is a no-op" + ); + + // The next invoke recompiles transparently and still works. + let r2 = rt.invoke(&cfg(), "on_file_uploaded", "inv2", &file_uploaded_input()); + assert!( + r2.outcome.is_ok(), + "recompile after eviction: {:?}", + r2.outcome + ); + // A long TTL never evicts a freshly-used module. + assert!( + !rt.evict_if_idle(Duration::from_secs(3600)), + "a fresh module must not be evicted" + ); +} + #[test] fn no_network() { let rt = PluginRuntime::new("com.example.net", fixture("net.wasm")); diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index f90b0ec7..98556dd4 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -1,6 +1,6 @@ use axum::{ Router, - extract::{Json, Multipart, Path, Query, State}, + extract::{DefaultBodyLimit, Json, Multipart, Path, Query, State}, http::{HeaderMap, StatusCode}, response::{ IntoResponse, @@ -70,7 +70,13 @@ pub fn admin_routes() -> Router> { .route("/photos/metadata/reextract", post(reextract_image_metadata)) // Plugin management .route("/plugins", get(list_plugins)) - .route("/plugins", post(install_plugin)) + // Install caps the request body at 32 MiB (overriding the global + // multi-GB upload limit) — a plugin bundle is small; the unpack also + // enforces a 64 MiB decompressed ceiling. + .route( + "/plugins", + post(install_plugin).layer(DefaultBodyLimit::max(32 * 1024 * 1024)), + ) .route("/plugins/{id}/enabled", put(set_plugin_enabled)) .route("/plugins/{id}", delete(delete_plugin)) // Plugin logs + per-plugin retention @@ -1502,6 +1508,9 @@ pub async fn list_plugins( admin_guard(&state, &headers).await?; let mgmt = plugin_mgmt(&state)?; let plugins: Vec = mgmt.list().into_iter().map(PluginInfoDto::from).collect(); + // `enabled` reports that the plugin *subsystem* is active (reaching here + // means it is — `plugin_mgmt` returns 503 otherwise, which the UI reads as + // the disabled state). Per-plugin enablement is each entry's own `enabled`. Ok(( StatusCode::OK, Json(serde_json::json!({ "enabled": true, "plugins": plugins })),