diff --git a/src/auth.rs b/src/auth.rs index b0d2165..225dbeb 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -5,7 +5,7 @@ //! The token is sent as `Authorization: Bearer `, or via the `?token=` //! query for SSE (EventSource cannot set headers). -use std::collections::HashSet; +use std::collections::HashMap; use std::sync::Mutex; use std::time::{Duration, Instant}; @@ -25,6 +25,8 @@ use crate::AppState; const MAX_FAILED_ATTEMPTS: u32 = 5; /// How long login stays locked after too many failures. const LOCKOUT: Duration = Duration::from_secs(30); +/// How long a session token stays valid after it is issued. +const SESSION_TTL: Duration = Duration::from_secs(7 * 24 * 60 * 60); /// Throttle state for the login endpoint (single admin account → process-global). #[derive(Default)] @@ -33,11 +35,11 @@ struct LoginGuard { locked_until: Option, } -/// Stores credentials + the set of active session tokens. +/// Stores credentials + active session tokens mapped to their expiry. pub struct Auth { user: String, pass: String, - sessions: Mutex>, + sessions: Mutex>, login_guard: Mutex, } @@ -46,7 +48,7 @@ impl Auth { Self { user, pass, - sessions: Mutex::new(HashSet::new()), + sessions: Mutex::new(HashMap::new()), login_guard: Mutex::new(LoginGuard::default()), } } @@ -87,12 +89,19 @@ impl Auth { fn issue(&self) -> String { *self.login_guard.lock().unwrap() = LoginGuard::default(); let token = Uuid::new_v4().to_string(); - self.sessions.lock().unwrap().insert(token.clone()); + let now = Instant::now(); + let mut sessions = self.sessions.lock().unwrap(); + // Drop expired tokens so the map stays bounded under repeated logins. + sessions.retain(|_, &mut expires| expires > now); + sessions.insert(token.clone(), now + SESSION_TTL); token } fn valid(&self, token: &str) -> bool { - self.sessions.lock().unwrap().contains(token) + match self.sessions.lock().unwrap().get(token) { + Some(&expires) => expires > Instant::now(), + None => false, + } } fn revoke(&self, token: &str) { @@ -191,6 +200,16 @@ mod tests { assert!(!ct_eq(b"secret", b"secre")); } + #[test] + fn session_issue_validate_revoke() { + let a = auth(); + let token = a.issue(); + assert!(a.valid(&token)); + assert!(!a.valid("not-a-token")); + a.revoke(&token); + assert!(!a.valid(&token)); + } + #[test] fn verify_accepts_only_exact_credentials() { let a = auth(); diff --git a/src/runtime/nodes/delay.rs b/src/runtime/nodes/delay.rs index 2cfd4a9..a09e8ef 100644 --- a/src/runtime/nodes/delay.rs +++ b/src/runtime/nodes/delay.rs @@ -7,7 +7,7 @@ use async_trait::async_trait; use serde::Deserialize; use serde_json::Value; -use crate::runtime::node::{Node, NodeCtx}; +use crate::runtime::node::{AbortOnDrop, Node, NodeCtx}; use crate::runtime::nodes::parse_config; fn default_mode() -> String { @@ -61,6 +61,8 @@ impl Node for DelayNode { // Default output (port 0) wires, for the spawned delay tasks. let outs = ctx.outs.first().cloned().unwrap_or_default(); let mut last: Option = None; + // In-flight delay tasks; aborted when the node stops (the Vec drops). + let mut pending: Vec = Vec::new(); while let Some(msg) = ctx.rx.recv().await { if self.mode == "rate" { @@ -69,13 +71,15 @@ impl Node for DelayNode { } } else { // Delay each message independently so the stream is not blocked. + pending.retain(|t| !t.0.is_finished()); // drop completed guards let outs = outs.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { tokio::time::sleep(interval).await; for out in &outs { let _ = out.send(msg.clone()).await; } }); + pending.push(AbortOnDrop(handle)); } } Ok(()) diff --git a/src/runtime/nodes/function.rs b/src/runtime/nodes/function.rs index 6f18ef0..f0c266d 100644 --- a/src/runtime/nodes/function.rs +++ b/src/runtime/nodes/function.rs @@ -28,6 +28,8 @@ use crate::runtime::nodes::parse_config; /// Cap on rhai operations per message — bounds runaway / `while true` scripts. const MAX_RHAI_OPS: u64 = 10_000_000; +/// Cap on JS loop iterations per message — bounds runaway `while (true)` scripts. +const MAX_JS_LOOP_ITERATIONS: u64 = 10_000_000; /// Fuel budget for one WASM `transform` call — bounds infinite loops in the module. const WASM_FUEL: u64 = 100_000_000; /// Cap on the WASM output buffer the host reads back (length is guest-controlled). @@ -120,6 +122,9 @@ fn eval_js(script: &str, msg: Msg) -> Result { use boa_engine::{Context, Source, js_string, property::Attribute}; let mut ctx = Context::default(); + // Bound runaway loops (the JS analog of rhai's operation cap). + ctx.runtime_limits_mut() + .set_loop_iteration_limit(MAX_JS_LOOP_ITERATIONS); let payload_js = boa_engine::JsValue::from_json(&msg.payload, &mut ctx).map_err(|e| e.to_string())?; ctx.register_global_property(js_string!("payload"), payload_js, Attribute::all()) @@ -308,4 +313,25 @@ mod tests { "expected an operation-limit error, got: {err}" ); } + + #[test] + fn js_transforms_payload() { + let out = Program::Js { + script: Arc::from("payload.n = payload.n + 1;"), + } + .eval(Msg::new(json!({ "n": 1 }))) + .expect("eval"); + assert_eq!(out.payload, json!({ "n": 2 })); + } + + #[test] + fn js_runaway_loop_is_bounded() { + // Without the loop-iteration cap this would never return. + let err = Program::Js { + script: Arc::from("while (true) {}"), + } + .eval(Msg::new(json!(null))) + .unwrap_err(); + assert!(!err.is_empty(), "expected a loop-limit error"); + } } diff --git a/src/runtime/nodes/link.rs b/src/runtime/nodes/link.rs index 5340b21..3ec82fc 100644 --- a/src/runtime/nodes/link.rs +++ b/src/runtime/nodes/link.rs @@ -17,20 +17,51 @@ use crate::model::Msg; use crate::runtime::node::{Node, NodeCtx}; use crate::runtime::nodes::parse_config; +/// A named link channel plus the number of nodes referencing it. +struct LinkChan { + tx: broadcast::Sender, + refs: usize, +} + /// Process-global registry of named link channels. -fn registry() -> &'static Mutex>> { - static LINKS: OnceLock>>> = OnceLock::new(); +fn registry() -> &'static Mutex> { + static LINKS: OnceLock>> = OnceLock::new(); LINKS.get_or_init(|| Mutex::new(HashMap::new())) } -/// The sender for a link name, creating it on first use. -fn channel_for(name: &str) -> broadcast::Sender { - registry() - .lock() - .unwrap() - .entry(name.to_string()) - .or_insert_with(|| broadcast::channel(64).0) - .clone() +/// Drops a link reference on drop, removing the channel when the last node +/// using the name stops — so the registry doesn't grow unbounded over deploys. +struct LinkGuard { + name: String, +} + +impl Drop for LinkGuard { + fn drop(&mut self) { + let mut reg = registry().lock().unwrap(); + if let Some(chan) = reg.get_mut(&self.name) { + chan.refs -= 1; + if chan.refs == 0 { + reg.remove(&self.name); + } + } + } +} + +/// Acquire the sender for a link name (creating it on first use) and a guard +/// that releases the reference when dropped. +fn acquire(name: &str) -> (broadcast::Sender, LinkGuard) { + let mut reg = registry().lock().unwrap(); + let chan = reg.entry(name.to_string()).or_insert_with(|| LinkChan { + tx: broadcast::channel(64).0, + refs: 0, + }); + chan.refs += 1; + ( + chan.tx.clone(), + LinkGuard { + name: name.to_string(), + }, + ) } #[derive(Debug, Default, Deserialize)] @@ -60,7 +91,8 @@ impl LinkOutNode { #[async_trait] impl Node for LinkInNode { async fn run(self: Box, ctx: NodeCtx) -> anyhow::Result<()> { - let mut rx = channel_for(&self.name).subscribe(); + let (tx, _guard) = acquire(&self.name); + let mut rx = tx.subscribe(); loop { match rx.recv().await { Ok(msg) => ctx.emit(msg).await, @@ -75,7 +107,7 @@ impl Node for LinkInNode { #[async_trait] impl Node for LinkOutNode { async fn run(self: Box, mut ctx: NodeCtx) -> anyhow::Result<()> { - let tx = channel_for(&self.name); + let (tx, _guard) = acquire(&self.name); while let Some(msg) = ctx.rx.recv().await { // Err = no link-in is listening on this name; ignore. let _ = tx.send(msg); @@ -89,10 +121,14 @@ mod tests { use super::*; use serde_json::json; + fn is_registered(name: &str) -> bool { + registry().lock().unwrap().contains_key(name) + } + #[test] fn same_name_shares_a_channel() { - let a = channel_for("link-test-share"); - let b = channel_for("link-test-share"); + let (a, _ga) = acquire("link-test-share"); + let (b, _gb) = acquire("link-test-share"); assert_eq!(a.receiver_count(), 0); let _rx = a.subscribe(); assert_eq!(b.receiver_count(), 1); // same underlying channel @@ -101,9 +137,20 @@ mod tests { #[tokio::test] async fn out_delivers_to_in() { // link-out → named channel → link-in subscriber. - let mut rx = channel_for("link-test-deliver").subscribe(); - let tx = channel_for("link-test-deliver"); + let (tx, _g) = acquire("link-test-deliver"); + let mut rx = tx.subscribe(); tx.send(Msg::new(json!("hi"))).unwrap(); assert_eq!(rx.recv().await.unwrap().payload, json!("hi")); } + + #[test] + fn channel_removed_when_last_ref_released() { + let (_tx1, g1) = acquire("link-test-refs"); + let (_tx2, g2) = acquire("link-test-refs"); + assert!(is_registered("link-test-refs")); + drop(g1); + assert!(is_registered("link-test-refs")); // still one holder + drop(g2); + assert!(!is_registered("link-test-refs")); // removed when unused + } } diff --git a/src/runtime/nodes/mqtt_shared.rs b/src/runtime/nodes/mqtt_shared.rs index 355ea34..a5b0e3e 100644 --- a/src/runtime/nodes/mqtt_shared.rs +++ b/src/runtime/nodes/mqtt_shared.rs @@ -104,12 +104,17 @@ pub fn get_or_create( node_id: String, build: impl FnOnce() -> Result<(Connection, Announce, Announce), String>, ) -> Result, String> { - let mut reg = registry().lock().unwrap(); - if let Some(shared) = reg.get(&key) { - shared.refs.fetch_add(1, Ordering::SeqCst); - return Ok(shared.clone()); + // Fast path: reuse an existing connection. + { + let reg = registry().lock().unwrap(); + if let Some(shared) = reg.get(&key) { + shared.refs.fetch_add(1, Ordering::SeqCst); + return Ok(shared.clone()); + } } + // Build outside the registry lock: constructing a TLS config can panic on + // unreadable platform certs, and that must not poison the global mutex. let ((client_box, poller), birth, close) = build()?; let client: Arc = Arc::from(client_box); let subs: Arc>> = Arc::new(Mutex::new(Vec::new())); @@ -121,7 +126,6 @@ pub fn get_or_create( events, node_id, )); - let shared = Arc::new(Shared { key: key.clone(), client, @@ -130,6 +134,19 @@ pub fn get_or_create( driver: Mutex::new(Some(driver)), close, }); + + // Insert, unless another task created the same connection while we built. + let mut reg = registry().lock().unwrap(); + if let Some(existing) = reg.get(&key) { + existing.refs.fetch_add(1, Ordering::SeqCst); + let existing = existing.clone(); + drop(reg); + // Discard our redundant connection. + if let Some(handle) = shared.driver.lock().unwrap().take() { + handle.abort(); + } + return Ok(existing); + } reg.insert(key, shared.clone()); Ok(shared) }