diff --git a/v2/crates/homecore-plugins/src/discovery.rs b/v2/crates/homecore-plugins/src/discovery.rs new file mode 100644 index 00000000..bfa8454d --- /dev/null +++ b/v2/crates/homecore-plugins/src/discovery.rs @@ -0,0 +1,270 @@ +//! Secure filesystem discovery for packaged WASM plugins. +//! +//! A plugin directory has exactly this shape: +//! +//! ```text +//! //manifest.json +//! +//! ``` +//! +//! Only immediate child directories of explicitly configured roots are +//! inspected. Symlinks are rejected and the module must be a plain filename +//! in the same package directory. This deliberately avoids recursive search, +//! implicit current-directory loading, and path traversal. + +use std::collections::BTreeSet; +use std::fs; +use std::path::{Component, Path, PathBuf}; + +use crate::{PluginError, PluginManifest}; + +/// Conservative input limits applied before allocating file contents. +#[derive(Debug, Clone, Copy)] +pub struct DiscoveryLimits { + pub max_plugins: usize, + pub max_manifest_bytes: u64, + pub max_module_bytes: u64, +} + +impl Default for DiscoveryLimits { + fn default() -> Self { + Self { + max_plugins: 128, + max_manifest_bytes: 256 * 1024, + max_module_bytes: 16 * 1024 * 1024, + } + } +} + +/// A validated package whose files are safe to read. +#[derive(Debug, Clone)] +pub struct DiscoveredPlugin { + pub manifest: PluginManifest, + pub package_dir: PathBuf, + pub manifest_path: PathBuf, + pub module_path: PathBuf, +} + +impl DiscoveredPlugin { + /// Read the module after re-checking its type and size. Re-checking closes + /// the common accidental replacement window between discovery and load; + /// signature verification remains the authoritative tamper gate. + pub fn read_module(&self, limits: DiscoveryLimits) -> Result, PluginError> { + bounded_regular_file(&self.module_path, limits.max_module_bytes, "WASM module") + } +} + +/// Discover packages in deterministic root/path order. +pub fn discover_plugins( + roots: &[PathBuf], + limits: DiscoveryLimits, +) -> Result, PluginError> { + if roots.is_empty() { + return Ok(Vec::new()); + } + if limits.max_plugins == 0 || limits.max_manifest_bytes == 0 || limits.max_module_bytes == 0 { + return Err(PluginError::ResourceLimit( + "discovery limits must all be greater than zero".into(), + )); + } + + let mut packages = BTreeSet::new(); + for configured_root in roots { + let metadata = fs::symlink_metadata(configured_root).map_err(|e| { + PluginError::Discovery(format!( + "cannot inspect configured plugin root {}: {e}", + configured_root.display() + )) + })?; + if metadata.file_type().is_symlink() || !metadata.is_dir() { + return Err(PluginError::Discovery(format!( + "configured plugin root must be a real directory, not a symlink: {}", + configured_root.display() + ))); + } + let root = fs::canonicalize(configured_root)?; + let entries = fs::read_dir(&root)?; + for entry in entries { + let entry = entry?; + let ty = entry.file_type()?; + if ty.is_symlink() { + return Err(PluginError::Discovery(format!( + "symlink found in plugin root: {}", + entry.path().display() + ))); + } + if ty.is_dir() && entry.path().join("manifest.json").exists() { + packages.insert(entry.path()); + } + } + } + + if packages.len() > limits.max_plugins { + return Err(PluginError::ResourceLimit(format!( + "discovered {} plugins, maximum is {}", + packages.len(), + limits.max_plugins + ))); + } + + let mut result = Vec::with_capacity(packages.len()); + let mut domains = BTreeSet::new(); + for package_dir in packages { + let package_dir = fs::canonicalize(package_dir)?; + let manifest_path = package_dir.join("manifest.json"); + let manifest_bytes = + bounded_regular_file(&manifest_path, limits.max_manifest_bytes, "manifest")?; + let manifest_text = std::str::from_utf8(&manifest_bytes).map_err(|e| { + PluginError::InvalidManifest(format!( + "{} is not UTF-8: {e}", + manifest_path.display() + )) + })?; + let manifest = PluginManifest::parse_json(manifest_text)?; + if !domains.insert(manifest.domain.clone()) { + return Err(PluginError::Discovery(format!( + "duplicate plugin domain `{}`", + manifest.domain + ))); + } + let module_name = manifest.wasm_module.as_deref().ok_or_else(|| { + PluginError::InvalidManifest(format!( + "WASM package `{}` has no wasm_module", + manifest.domain + )) + })?; + let relative = Path::new(module_name); + if relative.components().count() != 1 + || !matches!(relative.components().next(), Some(Component::Normal(_))) + || relative.extension().and_then(|v| v.to_str()) != Some("wasm") + { + return Err(PluginError::InvalidManifest(format!( + "plugin `{}` wasm_module must be a .wasm filename in its package directory", + manifest.domain + ))); + } + let module_path = package_dir.join(relative); + let metadata = fs::symlink_metadata(&module_path).map_err(|e| { + PluginError::Discovery(format!("cannot inspect {}: {e}", module_path.display())) + })?; + if metadata.file_type().is_symlink() || !metadata.is_file() { + return Err(PluginError::Discovery(format!( + "plugin module must be a regular non-symlink file: {}", + module_path.display() + ))); + } + if metadata.len() > limits.max_module_bytes { + return Err(PluginError::ResourceLimit(format!( + "{} is {} bytes; maximum is {}", + module_path.display(), + metadata.len(), + limits.max_module_bytes + ))); + } + result.push(DiscoveredPlugin { + manifest, + package_dir, + manifest_path, + module_path, + }); + } + Ok(result) +} + +fn bounded_regular_file(path: &Path, maximum: u64, kind: &str) -> Result, PluginError> { + let metadata = fs::symlink_metadata(path)?; + if metadata.file_type().is_symlink() || !metadata.is_file() { + return Err(PluginError::Discovery(format!( + "{kind} must be a regular non-symlink file: {}", + path.display() + ))); + } + if metadata.len() > maximum { + return Err(PluginError::ResourceLimit(format!( + "{} is {} bytes; maximum is {}", + path.display(), + metadata.len(), + maximum + ))); + } + let bytes = fs::read(path)?; + if bytes.len() as u64 > maximum { + return Err(PluginError::ResourceLimit(format!( + "{} grew beyond the {maximum}-byte limit while being read", + path.display() + ))); + } + Ok(bytes) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn root() -> PathBuf { + let path = std::env::temp_dir().join(format!( + "homecore-plugin-discovery-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + fs::create_dir(&path).unwrap(); + path + } + + fn package(root: &Path, directory: &str, domain: &str, module: &str) { + let dir = root.join(directory); + fs::create_dir(&dir).unwrap(); + fs::write( + dir.join("manifest.json"), + format!( + r#"{{"domain":"{domain}","name":"Test","version":"1","wasm_module":"{module}"}}"# + ), + ) + .unwrap(); + if !module.contains('/') && !module.contains('\\') && module.ends_with(".wasm") { + fs::write(dir.join(module), b"\0asm\x01\0\0\0").unwrap(); + } + } + + #[test] + fn discovery_is_sorted_and_only_reads_explicit_roots() { + let root = root(); + package(&root, "z-last", "zeta", "z.wasm"); + package(&root, "a-first", "alpha", "a.wasm"); + let found = discover_plugins(&[root.clone()], DiscoveryLimits::default()).unwrap(); + assert_eq!( + found.iter().map(|p| p.manifest.domain.as_str()).collect::>(), + ["alpha", "zeta"] + ); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn traversal_module_path_is_rejected() { + let root = root(); + package(&root, "bad", "bad", "../bad.wasm"); + let error = discover_plugins(&[root.clone()], DiscoveryLimits::default()).unwrap_err(); + assert!(matches!(error, PluginError::InvalidManifest(_))); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn size_limits_are_enforced_before_read() { + let root = root(); + package(&root, "large", "large", "large.wasm"); + let error = discover_plugins( + &[root.clone()], + DiscoveryLimits { + max_module_bytes: 4, + ..DiscoveryLimits::default() + }, + ) + .unwrap_err(); + assert!(matches!(error, PluginError::ResourceLimit(_))); + fs::remove_dir_all(root).unwrap(); + } +} diff --git a/v2/crates/homecore-plugins/src/error.rs b/v2/crates/homecore-plugins/src/error.rs index 468aa801..6c2cf20b 100644 --- a/v2/crates/homecore-plugins/src/error.rs +++ b/v2/crates/homecore-plugins/src/error.rs @@ -21,6 +21,14 @@ pub enum PluginError { #[error("runtime error: {0}")] RuntimeError(String), + /// Plugin discovery or filesystem layout is invalid. + #[error("plugin discovery failed: {0}")] + Discovery(String), + + /// A configured manifest, module, or ABI payload exceeds its limit. + #[error("plugin resource limit exceeded: {0}")] + ResourceLimit(String), + /// The plugin's `setup` hook returned an error. #[error("plugin setup failed: {0}")] SetupFailed(String), diff --git a/v2/crates/homecore-plugins/src/lib.rs b/v2/crates/homecore-plugins/src/lib.rs index c64d7d9f..a0ec6665 100644 --- a/v2/crates/homecore-plugins/src/lib.rs +++ b/v2/crates/homecore-plugins/src/lib.rs @@ -41,6 +41,7 @@ //! | `wasm3` | off | wasm3 interpreter runtime for constrained hardware (P3) | pub mod error; +pub mod discovery; pub mod host_abi; pub mod manifest; pub mod permissions; @@ -53,6 +54,7 @@ pub mod verify; pub mod wasmtime_runtime; pub use error::PluginError; +pub use discovery::{discover_plugins, DiscoveredPlugin, DiscoveryLimits}; pub use host_abi::{ConfigEntryJson, StateChangedEventJson}; pub use manifest::{IotClass, IntegrationType, PluginManifest}; pub use permissions::PermissionSet; diff --git a/v2/crates/homecore-plugins/src/plugin.rs b/v2/crates/homecore-plugins/src/plugin.rs index 634a72cd..4424cf32 100644 --- a/v2/crates/homecore-plugins/src/plugin.rs +++ b/v2/crates/homecore-plugins/src/plugin.rs @@ -11,10 +11,11 @@ use async_trait::async_trait; use homecore::HomeCore; use crate::error::PluginError; +use crate::StateChangedEventJson; /// Unique identifier for a loaded plugin — mirrors the `domain` field of /// the plugin's `PluginManifest` (e.g. `"mqtt"`, `"homecore_lights"`). -#[derive(Debug, Clone, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct PluginId(pub String); impl PluginId { @@ -52,6 +53,12 @@ pub trait HomeCorePlugin: Send + Sync + 'static { /// register its entities, services, and event subscriptions here. async fn setup(&self, hc: HomeCore) -> Result<(), PluginError>; + /// Receive a committed state change. The default is a no-op so built-in + /// plugins only opt into event work they need. + async fn state_changed(&self, _event: &StateChangedEventJson) -> Result<(), PluginError> { + Ok(()) + } + /// Called when the plugin is being removed from the registry. /// /// The plugin should clean up subscriptions and deregister its entities. diff --git a/v2/crates/homecore-plugins/src/registry.rs b/v2/crates/homecore-plugins/src/registry.rs index 28131d36..cff8a475 100644 --- a/v2/crates/homecore-plugins/src/registry.rs +++ b/v2/crates/homecore-plugins/src/registry.rs @@ -5,7 +5,7 @@ //! the `InProcessRuntime` (P1) for a `WasmtimeRuntime` (P2) without //! changing registry code. -use std::collections::HashMap; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use homecore::HomeCore; @@ -15,6 +15,7 @@ use crate::error::PluginError; use crate::manifest::PluginManifest; use crate::plugin::{HomeCorePlugin, PluginId}; use crate::runtime::{LoadedPlugin, PluginRuntime}; +use crate::StateChangedEventJson; /// Holds all loaded plugins keyed by `PluginId`. /// @@ -22,7 +23,8 @@ use crate::runtime::{LoadedPlugin, PluginRuntime}; /// unload) take an exclusive lock only while mutating the map. pub struct PluginRegistry { runtime: R, - plugins: RwLock>, + plugins: RwLock>, + loading: RwLock>, } impl PluginRegistry { @@ -30,7 +32,8 @@ impl PluginRegistry { pub fn new(runtime: R) -> Self { Self { runtime, - plugins: RwLock::new(HashMap::new()), + plugins: RwLock::new(BTreeMap::new()), + loading: RwLock::new(BTreeSet::new()), } } @@ -48,22 +51,31 @@ impl PluginRegistry { { let guard = self.plugins.read().await; - if guard.contains_key(&id) { + let mut loading = self.loading.write().await; + if guard.contains_key(&id) || !loading.insert(id.clone()) { return Err(PluginError::AlreadyLoaded(id.to_string())); } } - let loaded = self + let result = self .runtime .load(id.clone(), manifest, plugin) - .await?; - - loaded - .setup(hc) - .await - .map_err(|e| PluginError::SetupFailed(e.to_string()))?; - + .await; + let loaded = match result { + Ok(loaded) => loaded, + Err(error) => { + self.loading.write().await.remove(&id); + return Err(error); + } + }; + if let Err(error) = loaded.setup(hc).await { + // Best-effort rollback for partially initialized plugins. + let _ = loaded.unload().await; + self.loading.write().await.remove(&id); + return Err(PluginError::SetupFailed(error.to_string())); + } self.plugins.write().await.insert(id.clone(), loaded); + self.loading.write().await.remove(&id); Ok(id) } @@ -78,10 +90,11 @@ impl PluginRegistry { .ok_or_else(|| PluginError::NotFound(id.to_string()))? }; - loaded - .unload() - .await - .map_err(|e| PluginError::UnloadFailed(e.to_string()))?; + if let Err(error) = loaded.unload().await { + // Keep the handle reachable so teardown can be retried. + self.plugins.write().await.insert(id.clone(), loaded); + return Err(PluginError::UnloadFailed(error.to_string())); + } Ok(()) } @@ -99,4 +112,29 @@ impl PluginRegistry { pub async fn contains(&self, id: &PluginId) -> bool { self.plugins.read().await.contains_key(id) } + + /// Dispatch a state change in stable plugin-domain order. + pub async fn state_changed(&self, event: &StateChangedEventJson) -> Vec<(PluginId, PluginError)> { + let guard = self.plugins.read().await; + let mut errors = Vec::new(); + for (id, plugin) in guard.iter() { + if let Err(error) = plugin.state_changed(event).await { + errors.push((id.clone(), error)); + } + } + errors + } + + /// Tear down all plugins in reverse domain order. Failures are collected + /// while remaining plugins still receive teardown. + pub async fn shutdown(&self) -> Vec<(PluginId, PluginError)> { + let ids: Vec<_> = self.plugins.read().await.keys().cloned().rev().collect(); + let mut errors = Vec::new(); + for id in ids { + if let Err(error) = self.unload(&id).await { + errors.push((id, error)); + } + } + errors + } } diff --git a/v2/crates/homecore-plugins/src/runtime.rs b/v2/crates/homecore-plugins/src/runtime.rs index 9ff45ac9..ee00a47f 100644 --- a/v2/crates/homecore-plugins/src/runtime.rs +++ b/v2/crates/homecore-plugins/src/runtime.rs @@ -23,6 +23,7 @@ use homecore::HomeCore; use crate::error::PluginError; use crate::manifest::PluginManifest; use crate::plugin::{HomeCorePlugin, PluginId}; +use crate::StateChangedEventJson; /// A loaded plugin handle — returned by [`PluginRuntime::load`]. pub struct LoadedPlugin { @@ -42,6 +43,11 @@ impl LoadedPlugin { pub async fn unload(&self) -> Result<(), PluginError> { self.instance.unload().await } + + /// Dispatch a committed state change to this plugin. + pub async fn state_changed(&self, event: &StateChangedEventJson) -> Result<(), PluginError> { + self.instance.state_changed(event).await + } } /// Abstraction over the WASM (and native) plugin execution environment. diff --git a/v2/crates/homecore-plugins/src/wasmtime_runtime.rs b/v2/crates/homecore-plugins/src/wasmtime_runtime.rs index 3d4ca570..85615988 100644 --- a/v2/crates/homecore-plugins/src/wasmtime_runtime.rs +++ b/v2/crates/homecore-plugins/src/wasmtime_runtime.rs @@ -26,7 +26,7 @@ use std::sync::{Arc, Mutex}; use homecore::HomeCore; -use wasmtime::{Engine, Linker, Module, Store}; +use wasmtime::{Config, Engine, Linker, Module, Store, StoreLimits, StoreLimitsBuilder}; use crate::error::PluginError; use crate::host_abi::{LogLevel, StateChangedEventJson, MAX_ABI_BUFFER_BYTES}; @@ -34,6 +34,13 @@ use crate::manifest::PluginManifest; use crate::permissions::PermissionSet; use crate::verify::{verify_module, PluginPolicy}; +/// Hard ceiling for every module accepted by the runtime, including callers +/// that bypass filesystem discovery. +pub const MAX_WASM_MODULE_BYTES: usize = 16 * 1024 * 1024; +const MAX_SUBSCRIPTIONS: usize = 4096; +const MAX_LINEAR_MEMORY_BYTES: usize = 16 * 1024 * 1024; +const FUEL_PER_CALL: u64 = 10_000_000; + // ── Store data ───────────────────────────────────────────────────────────── /// Per-plugin state stored inside the Wasmtime [`Store`]. @@ -51,6 +58,7 @@ pub struct PluginStoreData { /// [`WasmtimeRuntime::load_plugin`] path installs the manifest's /// declared set. pub permissions: PermissionSet, + limits: StoreLimits, } // ── WasmtimeRuntime ──────────────────────────────────────────────────────── @@ -66,7 +74,10 @@ pub struct WasmtimeRuntime { impl WasmtimeRuntime { /// Create a new runtime with default Cranelift config. pub fn new() -> Result { - let engine = Engine::default(); + let mut config = Config::new(); + config.consume_fuel(true); + let engine = Engine::new(&config) + .map_err(|e| PluginError::RuntimeError(format!("Wasmtime engine: {e}")))?; Ok(Self { engine }) } @@ -83,6 +94,7 @@ impl WasmtimeRuntime { wasm_bytes: &[u8], hc: HomeCore, ) -> Result { + check_module_size(wasm_bytes)?; self.instantiate(wasm_bytes, hc, PermissionSet::allow_all()) } @@ -104,6 +116,7 @@ impl WasmtimeRuntime { hc: HomeCore, policy: &PluginPolicy, ) -> Result { + check_module_size(wasm_bytes)?; // P4: verify before instantiation. verify_module(manifest, wasm_bytes, policy)?; // P5: scope write authority to the manifest's declared permissions. @@ -128,8 +141,17 @@ impl WasmtimeRuntime { hc, subscriptions: Vec::new(), permissions, + limits: StoreLimitsBuilder::new() + .memory_size(MAX_LINEAR_MEMORY_BYTES) + .instances(1) + .memories(1) + .build(), }; let mut store = Store::new(&self.engine, store_data); + store.limiter(|data| &mut data.limits); + store + .set_fuel(FUEL_PER_CALL) + .map_err(|e| PluginError::RuntimeError(format!("set instantiation fuel: {e}")))?; let instance = linker .instantiate(&mut store, &module) @@ -179,6 +201,12 @@ fn register_hc_state_get( out_ptr: i32, out_cap: i32| -> i32 { + if out_ptr < 0 + || out_cap < 0 + || out_cap as usize > MAX_ABI_BUFFER_BYTES + { + return -1; + } // Phase 1: read the entity key from guest memory. let key: String = { let mem = match caller.get_export("memory") { @@ -216,7 +244,9 @@ fn register_hc_state_get( Some(wasmtime::Extern::Memory(m)) => m, _ => return -1, }; - let end = out_ptr as usize + json_bytes.len(); + let Some(end) = (out_ptr as usize).checked_add(json_bytes.len()) else { + return -1; + }; let out = match mem.data_mut(&mut caller).get_mut(out_ptr as usize..end) { Some(s) => s, None => return -1, @@ -331,7 +361,15 @@ fn register_hc_state_subscribe( None => return -1, } }; - caller.data_mut().subscriptions.push(eid); + if homecore::EntityId::parse(&eid).is_err() { + return -1; + } + if caller.data().subscriptions.len() >= MAX_SUBSCRIPTIONS { + return -2; + } + if !caller.data().subscriptions.contains(&eid) { + caller.data_mut().subscriptions.push(eid); + } 0 }, ) @@ -374,6 +412,7 @@ fn register_hc_log( /// /// The `Arc>` allows the handle to be `Clone` + `Send` while /// maintaining exclusive access for calls into the WASM module. +#[derive(Clone)] pub struct WasmPlugin { pub inner: Arc, wasmtime::Instance)>>, } @@ -394,6 +433,9 @@ impl WasmPlugin { .lock() .map_err(|e| PluginError::RuntimeError(format!("lock: {e}")))?; let (store, instance) = &mut *guard; + store + .set_fuel(FUEL_PER_CALL) + .map_err(|e| PluginError::RuntimeError(format!("set call fuel: {e}")))?; call_export_str(store, instance, "plugin_setup", config_entry_json) } @@ -409,20 +451,46 @@ impl WasmPlugin { .lock() .map_err(|e| PluginError::RuntimeError(format!("lock: {e}")))?; let (store, instance) = &mut *guard; + store + .set_fuel(FUEL_PER_CALL) + .map_err(|e| PluginError::RuntimeError(format!("set call fuel: {e}")))?; call_export_str(store, instance, "plugin_handle_state_changed", &json) } + + /// Call an optional `plugin_teardown() -> i32` export. Older modules that + /// do not export teardown remain compatible; new modules can release + /// guest-owned resources deterministically. + pub fn call_teardown(&self) -> Result { + let mut guard = self + .inner + .lock() + .map_err(|e| PluginError::RuntimeError(format!("lock: {e}")))?; + let (store, instance) = &mut *guard; + store + .set_fuel(FUEL_PER_CALL) + .map_err(|e| PluginError::RuntimeError(format!("set call fuel: {e}")))?; + let Some(func) = instance.get_func(&mut *store, "plugin_teardown") else { + return Ok(0); + }; + let func = func.typed::<(), i32>(&*store).map_err(|e| { + PluginError::RuntimeError(format!("plugin_teardown has invalid signature: {e}")) + })?; + func.call(&mut *store, ()) + .map_err(|e| PluginError::RuntimeError(format!("call plugin_teardown: {e}"))) + } } // ── Memory helpers ───────────────────────────────────────────────────────── /// Read a UTF-8 string from guest linear memory. fn read_str(mem: &[u8], ptr: i32, len: i32) -> Option<&str> { - if len < 0 || len as usize > MAX_ABI_BUFFER_BYTES { + if ptr < 0 || len < 0 || len as usize > MAX_ABI_BUFFER_BYTES { return None; } let ptr = ptr as usize; let len = len as usize; - let slice = mem.get(ptr..ptr + len)?; + let end = ptr.checked_add(len)?; + let slice = mem.get(ptr..end)?; std::str::from_utf8(slice).ok() } @@ -435,6 +503,13 @@ fn call_export_str( payload: &str, ) -> Result { let payload_bytes = payload.as_bytes().to_vec(); // owned copy avoids reborrow issues + if payload_bytes.len() > MAX_ABI_BUFFER_BYTES { + return Err(PluginError::ResourceLimit(format!( + "ABI payload is {} bytes; maximum is {}", + payload_bytes.len(), + MAX_ABI_BUFFER_BYTES + ))); + } let payload_len = payload_bytes.len() as i32; // 1. Allocate guest buffer. @@ -444,15 +519,23 @@ fn call_export_str( let ptr = alloc .call(&mut *store, payload_len) .map_err(|e| PluginError::RuntimeError(format!("call alloc: {e}")))?; + if ptr < 0 { + return Err(PluginError::RuntimeError( + "guest alloc returned a negative pointer".into(), + )); + } // 2. Write payload into guest memory. { let mem = instance .get_memory(&mut *store, "memory") .ok_or_else(|| PluginError::RuntimeError("no memory export".into()))?; + let end = (ptr as usize) + .checked_add(payload_bytes.len()) + .ok_or_else(|| PluginError::RuntimeError("guest allocation overflow".into()))?; let guest_slice = mem .data_mut(&mut *store) - .get_mut(ptr as usize..ptr as usize + payload_bytes.len()) + .get_mut(ptr as usize..end) .ok_or_else(|| PluginError::RuntimeError("guest memory OOB".into()))?; guest_slice.copy_from_slice(&payload_bytes); } @@ -476,6 +559,17 @@ fn call_export_str( Ok(result) } +fn check_module_size(wasm_bytes: &[u8]) -> Result<(), PluginError> { + if wasm_bytes.len() > MAX_WASM_MODULE_BYTES { + return Err(PluginError::ResourceLimit(format!( + "WASM module is {} bytes; maximum is {}", + wasm_bytes.len(), + MAX_WASM_MODULE_BYTES + ))); + } + Ok(()) +} + // ── Unit tests (using inline WAT) ────────────────────────────────────────── #[cfg(test)] diff --git a/v2/crates/homecore-server/Cargo.toml b/v2/crates/homecore-server/Cargo.toml index 5a483acd..0bb62d61 100644 --- a/v2/crates/homecore-server/Cargo.toml +++ b/v2/crates/homecore-server/Cargo.toml @@ -23,7 +23,7 @@ path = "src/main.rs" # The 8 HOMECORE crates this binary integrates homecore = { path = "../homecore", version = "0.1.0-alpha.0" } homecore-api = { path = "../homecore-api", version = "0.1.0-alpha.0" } -homecore-plugins = { path = "../homecore-plugins", version = "0.1.0-alpha.0", optional = true } +homecore-plugins = { path = "../homecore-plugins", version = "0.1.0-alpha.0" } homecore-recorder = { path = "../homecore-recorder", version = "0.1.0-alpha.0" } # Reuse version-gated registry envelope parsing in the isolated restore module. homecore-migrate = { path = "../homecore-migrate", version = "0.1.0-alpha.0" } @@ -65,4 +65,4 @@ default = [] # Pull in ruvector-backed semantic memory. ruvector = ["homecore-recorder/ruvector"] # Pull in real Wasmtime plugin runtime (vs InProcessRuntime). -wasmtime = ["dep:homecore-plugins", "homecore-plugins/wasmtime"] +wasmtime = ["homecore-plugins/wasmtime"] diff --git a/v2/crates/homecore-server/src/plugins.rs b/v2/crates/homecore-server/src/plugins.rs new file mode 100644 index 00000000..5693921f --- /dev/null +++ b/v2/crates/homecore-server/src/plugins.rs @@ -0,0 +1,215 @@ +//! Server ownership for plugin discovery, setup, event dispatch, and teardown. +//! +//! Native Rust plugins are never discovered from disk. They must be compiled +//! into this binary and exposed through [`NativePluginRegistration`]. External +//! packages are WebAssembly only and require the `wasmtime` feature. + +use std::path::PathBuf; +use std::sync::Arc; + +use anyhow::{Context as _, Result}; +use homecore::HomeCore; +use homecore_plugins::{ + DiscoveryLimits, HomeCorePlugin, InProcessRuntime, PluginError, PluginManifest, + PluginRegistry, StateChangedEventJson, +}; +use tokio::task::JoinHandle; +use tracing::{error, info, warn}; + +#[cfg(feature = "wasmtime")] +use homecore_plugins::{ + discover_plugins, ConfigEntryJson, PluginId, PluginPolicy, WasmPlugin, WasmtimeRuntime, +}; + +/// Factory entry for a trusted plugin linked into the server at compile time. +pub struct NativePluginRegistration { + pub create: + fn() -> Result<(PluginManifest, Arc), PluginError>, +} + +/// This is intentionally empty in the generic server. Appliance builds can +/// add first-party registrations here; arbitrary Rust cdylibs are unsupported. +const NATIVE_PLUGINS: &[NativePluginRegistration] = &[]; + +#[derive(Debug, Clone)] +pub struct PluginConfig { + pub directories: Vec, + pub trusted_publishers: Vec, + pub allow_unsigned: bool, + pub limits: DiscoveryLimits, +} + +pub struct ServerPlugins { + native: Arc>, + dispatcher: JoinHandle<()>, + #[cfg(feature = "wasmtime")] + wasm: Vec<(PluginId, WasmPlugin)>, +} + +impl ServerPlugins { + /// Start the complete plugin subsystem. Any configured package that fails + /// discovery, trust verification, compilation, or setup aborts startup. + pub async fn start(hc: HomeCore, config: PluginConfig) -> Result { + let native = Arc::new(PluginRegistry::new(InProcessRuntime)); + for registration in NATIVE_PLUGINS { + let (manifest, plugin) = (registration.create)() + .context("compiled-in native plugin factory failed")?; + native + .load(manifest, plugin, hc.clone()) + .await + .context("compiled-in native plugin setup failed")?; + } + + #[cfg(not(feature = "wasmtime"))] + if !config.directories.is_empty() { + anyhow::bail!( + "plugin directories were configured, but homecore-server was built without \ + --features wasmtime" + ); + } + #[cfg(not(feature = "wasmtime"))] + if config.allow_unsigned || !config.trusted_publishers.is_empty() { + anyhow::bail!( + "plugin trust options were configured, but homecore-server was built without \ + --features wasmtime" + ); + } + + #[cfg(feature = "wasmtime")] + let wasm = load_wasm_plugins(&hc, &config).await?; + + let mut receiver = hc.states().subscribe(); + let dispatch_native = native.clone(); + #[cfg(feature = "wasmtime")] + let dispatch_wasm = wasm.clone(); + let dispatcher = tokio::spawn(async move { + loop { + let change = match receiver.recv().await { + Ok(change) => change, + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + warn!(skipped, "plugin state dispatcher lagged; events were dropped"); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + }; + let event = StateChangedEventJson::state_changed( + change.entity_id.as_str(), + change.new_state.as_ref().map(|state| state.state.as_str()), + change + .new_state + .as_ref() + .map(|state| state.attributes.clone()) + .unwrap_or_else(|| serde_json::json!({})), + ); + for (id, error) in dispatch_native.state_changed(&event).await { + error!(plugin = %id, %error, "native plugin state_changed failed"); + } + #[cfg(feature = "wasmtime")] + for (id, plugin) in &dispatch_wasm { + if !plugin.subscriptions().iter().any(|entity| entity == change.entity_id.as_str()) + { + continue; + } + let plugin = plugin.clone(); + let id = id.clone(); + let event = StateChangedEventJson::state_changed( + &event.entity_id, + event.new_state.as_deref(), + event.attributes.clone(), + ); + match tokio::task::spawn_blocking(move || plugin.call_state_changed(&event)).await + { + Ok(Ok(0)) => {} + Ok(Ok(code)) => error!(plugin = %id, code, "WASM state_changed returned failure"), + Ok(Err(error)) => error!(plugin = %id, %error, "WASM state_changed trapped"), + Err(error) => error!(plugin = %id, %error, "WASM dispatch task failed"), + } + } + } + }); + + info!( + native = NATIVE_PLUGINS.len(), + #[cfg(feature = "wasmtime")] + wasm = wasm.len(), + "plugin subsystem started" + ); + Ok(Self { + native, + dispatcher, + #[cfg(feature = "wasmtime")] + wasm, + }) + } + + /// Stop dispatch first, then tear down WASM and native plugins in reverse + /// deterministic order. All teardown failures are reported. + pub async fn shutdown(self) { + self.dispatcher.abort(); + let _ = self.dispatcher.await; + #[cfg(feature = "wasmtime")] + for (id, plugin) in self.wasm.into_iter().rev() { + match tokio::task::spawn_blocking(move || plugin.call_teardown()).await { + Ok(Ok(0)) => {} + Ok(Ok(code)) => error!(plugin = %id, code, "WASM teardown returned failure"), + Ok(Err(error)) => error!(plugin = %id, %error, "WASM teardown trapped"), + Err(error) => error!(plugin = %id, %error, "WASM teardown task failed"), + } + } + for (id, error) in self.native.shutdown().await { + error!(plugin = %id, %error, "native plugin teardown failed"); + } + } +} + +#[cfg(feature = "wasmtime")] +async fn load_wasm_plugins( + hc: &HomeCore, + config: &PluginConfig, +) -> Result> { + let policy = if config.allow_unsigned { + warn!("INSECURE plugin policy enabled: unsigned plugins may execute"); + PluginPolicy::AllowUnsigned + } else { + let keys: Vec<_> = config.trusted_publishers.iter().map(String::as_str).collect(); + PluginPolicy::trusted(&keys).context("invalid trusted plugin publisher key")? + }; + let discovered = discover_plugins(&config.directories, config.limits) + .context("plugin discovery failed")?; + let runtime = WasmtimeRuntime::new().context("Wasmtime initialization failed")?; + let mut loaded: Vec<(PluginId, WasmPlugin)> = Vec::with_capacity(discovered.len()); + for package in discovered { + let id = PluginId::new(&package.manifest.domain); + let bytes = package + .read_module(config.limits) + .with_context(|| format!("failed reading plugin `{id}`"))?; + let plugin = runtime + .load_plugin(&package.manifest, &bytes, hc.clone(), &policy) + .with_context(|| format!("plugin `{id}` rejected before setup"))?; + let config_entry = serde_json::to_string(&ConfigEntryJson::bootstrap(id.as_str()))?; + let setup_plugin = plugin.clone(); + let result = tokio::task::spawn_blocking(move || setup_plugin.call_setup(&config_entry)) + .await + .with_context(|| format!("plugin `{id}` setup task failed"))? + .with_context(|| format!("plugin `{id}` setup trapped"))?; + if result != 0 { + let _ = plugin.call_teardown(); + teardown_wasm(loaded).await; + anyhow::bail!("plugin `{id}` setup returned failure code {result}"); + } + info!( + plugin = %id, + package = %package.package_dir.display(), + "signed WASM plugin loaded" + ); + loaded.push((id, plugin)); + } + Ok(loaded) +} + +#[cfg(feature = "wasmtime")] +async fn teardown_wasm(plugins: Vec<(PluginId, WasmPlugin)>) { + for (_, plugin) in plugins.into_iter().rev() { + let _ = tokio::task::spawn_blocking(move || plugin.call_teardown()).await; + } +}