security(wasm): bound sandbox resources and remove host-side panic/DoS vectors

Hardens the WASM host runtime against a hostile or compromised plugin
(area: WASM sandbox limits).

- Panic vectors: guest-controlled pointer/length slice reads use
  saturating_add so a negative/oversized len fails the existing bounds
  check instead of panicking on an inverted range. Cache and rate-limiter
  Instant/Duration arithmetic is overflow/underflow-safe (checked/
  saturating + duration_since), including the cleanup paths that
  underflow early in process uptime.
- Memory bounds: the response cache caps its entry count (evicting
  expired-then-soonest-to-expire); the rate limiter clamps the
  plugin-supplied quota and window and caps the partition table, failing
  closed when saturated with active keys.
- Upstream body cap: the buffered plugin HTTP-call path reads at most
  BARBACANE_MAX_UPSTREAM_RESPONSE_BYTES (default 16 MiB) via a chunked,
  content-length-aware read, bounding host memory.
- Wall-clock backstop: epoch interruption with a background ticker traps
  a guest that runs past its time budget, complementing fuel-based CPU
  limiting.
- Broker hardening: Kafka/NATS connections enforce the SSRF guard
  (honoring BARBACANE_ALLOW_INTERNAL_EGRESS), connect/publish timeouts,
  and bounded connection caches. Runtime build errors propagate instead
  of panicking.

Docs and CHANGELOG updated for the new env var and broker egress default.

Author: ndreno

CHANGELOG.md

@@ -15,19 +15,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **security**: WASM plugin capability enforcement — plugins may only import host functions covered by the capabilities declared in `plugin.toml`.
 - **security**: `jwt-auth` performs real signature verification via the host `verify_signature` capability (inline JWK).
 - **security**: a security testing framework — adversarial integration suite (`crates/barbacane-test/tests/security/`) and `cargo-fuzz` targets (`fuzz/`).
+- **security**: WASM sandbox resource limits — buffered plugin HTTP responses are capped (`BARBACANE_MAX_UPSTREAM_RESPONSE_BYTES`, default 16 MiB); the host cache and rate limiter bound their entry/partition counts and clamp plugin-supplied TTL/window/quota; a wall-clock epoch deadline backstops fuel-based CPU limiting.
 - **docs**: [Configuration & environment variables](reference/configuration.md) reference.
 
 ### Changed (breaking, secure-by-default)
 
 - The control plane refuses to start without `BARBACANE_CONTROL_ADMIN_TOKEN`, and all API routes (except `/health` and the data-plane WebSocket) require the bearer token.
 - `file://` secret references require `BARBACANE_SECRETS_DIR` and are confined to it.
 - MCP requires a valid session for non-`initialize` requests.
-- Plugin HTTP egress to internal/metadata addresses is blocked by default.
+- Plugin egress to internal/metadata addresses is blocked by default — this now covers Kafka/NATS broker connections in addition to plugin HTTP calls.
 
 ### Fixed
 
 - **security**: fail-open middleware short-circuit downgrade in the WASM chain.
 - **security**: panic on hostile `x-request-id` / `traceparent`; unbounded Prometheus path-label cardinality on unmatched routes.
+- **security**: panic vectors in WASM host functions — guest-controlled pointer/length slice reads now use saturating arithmetic, and cache/rate-limiter time arithmetic is overflow/underflow-safe.
 - **deps**: bump `anyhow` to 1.0.103 (RUSTSEC-2026-0190).
 
 ## [0.7.0] - 2026-05-05

crates/barbacane-wasm/src/broker.rs

@@ -24,6 +24,48 @@ pub enum BrokerError {
 
     #[error("timeout")]
     Timeout,
+
+    #[error("broker target blocked by SSRF policy: {0}")]
+    Blocked(String),
+}
+
+/// Split a broker address into its host and port, stripping any `scheme://`
+/// prefix and IPv6 brackets. Used to feed the SSRF guard. Falls back to
+/// `default_port` when no port is present.
+pub(crate) fn split_host_port(addr: &str, default_port: u16) -> (String, u16) {
+    // Drop a leading scheme such as `nats://` or `tls://`.
+    let addr = addr
+        .split_once("://")
+        .map(|(_, rest)| rest)
+        .unwrap_or(addr)
+        .trim();
+    // Drop any path/query that may follow the authority.
+    let authority = addr
+        .split(['/', '?'])
+        .next()
+        .unwrap_or(addr)
+        .trim_end_matches('.');
+
+    // Bracketed IPv6: [::1]:4222 or [::1]
+    if let Some(rest) = authority.strip_prefix('[') {
+        if let Some((host, after)) = rest.split_once(']') {
+            let port = after
+                .strip_prefix(':')
+                .and_then(|p| p.parse::<u16>().ok())
+                .unwrap_or(default_port);
+            return (host.to_string(), port);
+        }
+    }
+
+    // host:port — but only treat the last colon as a port separator when what
+    // follows parses as a port (avoids mangling a bare unbracketed IPv6).
+    if let Some((host, port_str)) = authority.rsplit_once(':') {
+        if let Ok(port) = port_str.parse::<u16>() {
+            return (host.to_string(), port);
+        }
+    }
+
+    (authority.to_string(), default_port)
 }
 
 /// A message to publish to a broker.
@@ -215,6 +257,30 @@ mod tests {
         assert_eq!(deserialized.offset, Some(42));
     }
 
+    #[test]
+    fn split_host_port_variants() {
+        assert_eq!(split_host_port("kafka:9092", 9092), ("kafka".into(), 9092));
+        assert_eq!(
+            split_host_port("nats://example.com:4222", 4222),
+            ("example.com".into(), 4222)
+        );
+        assert_eq!(
+            split_host_port("broker.internal", 9092),
+            ("broker.internal".into(), 9092)
+        );
+        assert_eq!(split_host_port("[::1]:4222", 4222), ("::1".into(), 4222));
+        assert_eq!(split_host_port("[fe80::1]", 4222), ("fe80::1".into(), 4222));
+        assert_eq!(
+            split_host_port("tls://10.0.0.1:9093/path", 9092),
+            ("10.0.0.1".into(), 9093)
+        );
+        // Bare IPv6 without brackets: no colon is treated as a port.
+        assert_eq!(
+            split_host_port("169.254.169.254:9092", 9092),
+            ("169.254.169.254".into(), 9092)
+        );
+    }
+
     #[test]
     fn broker_error_display() {
         assert_eq!(

crates/barbacane-wasm/src/cache.rs

@@ -8,6 +8,16 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+/// Upper bound on an entry's TTL. The TTL is plugin-controlled, so it is clamped
+/// before being fed into `Instant + Duration` arithmetic to avoid an
+/// overflow panic on an absurd value (1 year is well beyond any sane cache TTL).
+const MAX_TTL_SECS: u64 = 366 * 24 * 60 * 60;
+
+/// Upper bound on the number of cached entries. A plugin can write arbitrary
+/// keys, so the table is capped to bound host memory; inserts past the cap evict
+/// expired entries first, then the entry closest to expiry.
+const MAX_CACHE_ENTRIES: usize = 10_000;
+
 /// A cached response entry.
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct CacheEntry {
@@ -97,8 +107,8 @@ impl ResponseCache {
 
                 let mut entry = internal.entry.clone();
                 entry.metadata = Some(CacheMetadata {
-                    created_at: now_unix - internal.created_at.elapsed().as_secs(),
-                    expires_at: now_unix + ttl_remaining,
+                    created_at: now_unix.saturating_sub(internal.created_at.elapsed().as_secs()),
+                    expires_at: now_unix.saturating_add(ttl_remaining),
                     ttl_remaining,
                 });
 
@@ -119,7 +129,12 @@ impl ResponseCache {
     /// Set a cache entry with TTL.
     pub fn set(&self, key: &str, entry: CacheEntry, ttl_secs: u64) {
         let now = Instant::now();
-        let expires_at = now + Duration::from_secs(ttl_secs);
+        // Clamp the plugin-controlled TTL and use checked arithmetic so a huge
+        // value can't overflow-panic `Instant + Duration`.
+        let ttl_secs = ttl_secs.min(MAX_TTL_SECS);
+        let expires_at = now
+            .checked_add(Duration::from_secs(ttl_secs))
+            .unwrap_or(now);
 
         let internal = InternalEntry {
             entry,
@@ -129,6 +144,20 @@ impl ResponseCache {
         };
 
         let mut entries = self.entries.write();
+        if entries.len() >= MAX_CACHE_ENTRIES && !entries.contains_key(key) {
+            // Make room: drop expired entries first.
+            entries.retain(|_, v| v.expires_at > now);
+            // Still full of live entries? Evict the one closest to expiry.
+            if entries.len() >= MAX_CACHE_ENTRIES {
+                if let Some(victim) = entries
+                    .iter()
+                    .min_by_key(|(_, v)| v.expires_at)
+                    .map(|(k, _)| k.clone())
+                {
+                    entries.remove(&victim);
+                }
+            }
+        }
         entries.insert(key.to_string(), internal);
     }
 
@@ -221,6 +250,26 @@ mod tests {
         assert_eq!(cached.body, Some(b"test body".to_vec()));
     }
 
+    #[test]
+    fn cache_entry_count_is_bounded() {
+        let cache = ResponseCache::new();
+        let entry = CacheEntry {
+            status: 200,
+            headers: HashMap::new(),
+            body: None,
+            metadata: None,
+        };
+        // Write more distinct keys than the cap, all with a live TTL.
+        for i in 0..(MAX_CACHE_ENTRIES + 100) {
+            cache.set(&format!("key-{i}"), entry.clone(), 3600);
+        }
+        assert!(
+            cache.stats().total_entries <= MAX_CACHE_ENTRIES,
+            "cache grew past the cap: {}",
+            cache.stats().total_entries
+        );
+    }
+
     #[test]
     fn test_cache_invalidate() {
         let cache = ResponseCache::new();

crates/barbacane-wasm/src/engine.rs

@@ -3,11 +3,21 @@
 //! This module provides the core wasmtime engine with settings optimized
 //! for the Barbacane plugin runtime.
 
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread::JoinHandle;
+use std::time::Duration;
+
 use wasmtime::{Config, Engine, Module, OptLevel};
 
 use crate::error::WasmError;
 use crate::limits::PluginLimits;
 
+/// Wall-clock granularity of the epoch ticker. The per-call epoch deadline is
+/// expressed as a count of these ticks, so it bounds the slack between a
+/// plugin's configured execution budget and when it is actually interrupted.
+pub(crate) const EPOCH_TICK: Duration = Duration::from_millis(1);
+
 /// A compiled WASM module ready for instantiation.
 #[derive(Clone)]
 pub struct CompiledModule {
@@ -31,6 +41,19 @@ impl CompiledModule {
 pub struct WasmEngine {
     engine: Engine,
     limits: PluginLimits,
+    /// Signals the epoch ticker thread to stop on drop.
+    epoch_stop: Arc<AtomicBool>,
+    /// Handle to the background epoch ticker thread.
+    epoch_ticker: Option<JoinHandle<()>>,
+}
+
+impl Drop for WasmEngine {
+    fn drop(&mut self) {
+        self.epoch_stop.store(true, Ordering::Relaxed);
+        if let Some(handle) = self.epoch_ticker.take() {
+            let _ = handle.join();
+        }
+    }
 }
 
 impl WasmEngine {
@@ -49,6 +72,13 @@ impl WasmEngine {
         // Enable fuel consumption for execution time limiting
         config.consume_fuel(true);
 
+        // Enable epoch interruption as a wall-clock backstop. Fuel bounds the
+        // number of instructions executed, but not wall-clock time (e.g. on a
+        // slow host, or if fuel is miscalibrated). A background thread ticks the
+        // engine epoch and each call sets a deadline, trapping a guest that runs
+        // past its time budget.
+        config.epoch_interruption(true);
+
         // Configure memory settings
         config.max_wasm_stack(limits.max_stack_bytes);
 
@@ -66,7 +96,29 @@ impl WasmEngine {
 
         let engine = Engine::new(&config).map_err(|e| WasmError::EngineCreation(e.to_string()))?;
 
-        Ok(Self { engine, limits })
+        // Spawn the epoch ticker: it increments the engine epoch every
+        // `EPOCH_TICK` so per-call epoch deadlines fire on a wall-clock basis.
+        let epoch_stop = Arc::new(AtomicBool::new(false));
+        let epoch_ticker = {
+            let engine = engine.clone();
+            let stop = Arc::clone(&epoch_stop);
+            std::thread::Builder::new()
+                .name("wasm-epoch-ticker".into())
+                .spawn(move || {
+                    while !stop.load(Ordering::Relaxed) {
+                        std::thread::sleep(EPOCH_TICK);
+                        engine.increment_epoch();
+                    }
+                })
+                .map_err(|e| WasmError::EngineCreation(e.to_string()))?
+        };
+
+        Ok(Self {
+            engine,
+            limits,
+            epoch_stop,
+            epoch_ticker: Some(epoch_ticker),
+        })
     }
 
     /// Get a reference to the underlying wasmtime engine.