fix(hu-union): harden wasm plugin loading against untrusted manifests

- sanitize filename-derived plugin stem to prevent work-dir path traversal
- bound the rate-tracker channel to cap host memory on high-rate topics
- gate declared session opening behind the OpenSession permission
- spawn an OS-thread epoch ticker when no Tokio runtime is present so
  runaway plugins stay preemptible

Author: YuanYuYuan

crates/hiroz-union/src/plugin/wasm/mod.rs

@@ -225,14 +225,29 @@ fn shared_wasm_engine() -> Result<Engine> {
     // engine that wins the `get_or_init` race gets its ticker spawned.
     let candidate = configured_wasm_engine()?;
     let engine = SHARED_WASM_ENGINE.get_or_init(|| {
+        let ticker_engine = candidate.clone();
         if let Ok(handle) = tokio::runtime::Handle::try_current() {
-            let ticker_engine = candidate.clone();
             handle.spawn(async move {
                 loop {
                     tokio::time::sleep(Duration::from_millis(100)).await;
                     ticker_engine.increment_epoch();
                 }
             });
+        } else {
+            // No Tokio runtime in this context. Epoch interruption would be
+            // silently disabled (the epoch would never advance, so a runaway
+            // plugin could not be preempted). Fall back to a dedicated OS
+            // thread that increments the epoch on the same cadence, so
+            // preemption keeps working regardless of the caller's runtime.
+            std::thread::Builder::new()
+                .name("hu-wasm-epoch".into())
+                .spawn(move || {
+                    loop {
+                        std::thread::sleep(Duration::from_millis(100));
+                        ticker_engine.increment_epoch();
+                    }
+                })
+                .ok();
         }
         candidate
     });
@@ -349,12 +364,36 @@ fn iter_wasm_files() -> impl Iterator<Item = PathBuf> {
         .filter(|p| p.extension().and_then(|e| e.to_str()) == Some("wasm"))
 }
 
+/// Sanitize a filename-derived plugin stem for safe use as a single path
+/// segment. The stem comes straight from an on-disk filename, so a crafted
+/// name like `hu-..\.wasm` would otherwise yield a stem of `..` and let the
+/// per-plugin work dir escape its base (path traversal). Keep only a safe
+/// `[A-Za-z0-9_-]` set (path separators and dots become `_`), and never let
+/// the result be empty or a `.`/`..` traversal component.
+fn sanitize_plugin_stem(plugin_stem: &str) -> String {
+    let cleaned: String = plugin_stem
+        .chars()
+        .map(|c| {
+            if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
+                c
+            } else {
+                '_'
+            }
+        })
+        .collect();
+    if cleaned.is_empty() || cleaned.chars().all(|c| c == '.') {
+        "unknown".to_string()
+    } else {
+        cleaned
+    }
+}
+
 fn plugin_work_dir(plugin_stem: &str) -> PathBuf {
     dirs::data_local_dir()
         .unwrap_or_else(|| PathBuf::from("."))
         .join("hu")
         .join("plugin-work")
-        .join(plugin_stem)
+        .join(sanitize_plugin_stem(plugin_stem))
 }
 
 type StateAndStore = (
@@ -501,6 +540,22 @@ fn open_declared_sessions(
     store: &mut Store<PluginState>,
     manifest: &hu::plugin::types::PluginManifest,
 ) -> Result<()> {
+    // The manifest is untrusted: opening a session triggers an outbound Zenoh
+    // connection, so it must be gated exactly like the runtime `open_session`
+    // host call (see host/transport.rs). Refuse to open *any* declared session
+    // unless the plugin was granted `OpenSession` — otherwise a plugin could
+    // trigger network connections it never declared a permission for.
+    if !manifest.sessions.is_empty()
+        && !manifest
+            .required_permissions
+            .contains(&hu::plugin::types::Permission::OpenSession)
+    {
+        anyhow::bail!(
+            "plugin declares {} session(s) but did not request the OpenSession permission",
+            manifest.sessions.len()
+        );
+    }
+
     for req in &manifest.sessions {
         let name = req.name.clone();
         let endpoint = req.endpoint.clone();

crates/hiroz-union/src/plugin/wasm/state.rs

@@ -13,6 +13,10 @@ use crate::core::engine::CoreEngine;
 use super::host::hu;
 use hu::plugin::types::Permission;
 
+/// Upper bound on buffered rate-tracker arrivals per topic. Caps host memory
+/// on high-rate topics; anything beyond this window is stale and dropped.
+const RATE_TRACKER_CHANNEL_CAP: usize = 16_384;
+
 // ─── Subscription tracking (ros interface) ────────────────────────────────────
 
 pub(crate) struct SubscriptionData {
@@ -162,7 +166,12 @@ impl PluginState {
         let topic_stripped = topic.trim_start_matches('/').to_string();
         let ke = format!("{domain_id}/{topic_stripped}/**");
         let session = self.engine.session.clone();
-        let (tx, rx) = flume::unbounded::<(Instant, usize)>();
+        // Bounded so a high-rate topic (or infrequent `measure` calls that drain
+        // the tracker slowly) can't grow host memory without bound / OOM. The
+        // window is trimmed on every measurement anyway, so a backlog beyond this
+        // cap is already stale; best-effort send drops the newest arrival when
+        // full rather than blocking the subscriber task.
+        let (tx, rx) = flume::bounded::<(Instant, usize)>(RATE_TRACKER_CHANNEL_CAP);
         let ke_clone = ke.clone();
         let handle = tokio::spawn(async move {
             let sub = match session.declare_subscriber(&ke_clone).await {
@@ -174,10 +183,13 @@ impl PluginState {
             };
             while let Ok(sample) = sub.recv_async().await {
                 let size = sample.payload().to_bytes().len();
-                // Stop once the tracker (receiver) is gone — otherwise this
-                // loop spins forever after the RateTrackerData is dropped.
-                if tx.send((Instant::now(), size)).is_err() {
-                    break;
+                match tx.try_send((Instant::now(), size)) {
+                    Ok(()) => {}
+                    // Channel full: drop this arrival (best-effort) and keep going.
+                    Err(flume::TrySendError::Full(_)) => {}
+                    // Stop once the tracker (receiver) is gone — otherwise this
+                    // loop spins forever after the RateTrackerData is dropped.
+                    Err(flume::TrySendError::Disconnected(_)) => break,
                 }
             }
         });