diff --git a/net-guardia/src/core/soar/actions.rs b/net-guardia/src/core/soar/actions.rs index 18e1861..6014f10 100644 --- a/net-guardia/src/core/soar/actions.rs +++ b/net-guardia/src/core/soar/actions.rs @@ -130,26 +130,15 @@ impl SoarEngine { ) -> Result { let ttl_secs = action.params.get("ttl_secs").and_then(|v| v.as_u64()).unwrap_or(1800); - // Validate TTL (runtime-configurable via DB) - let max_ttl: u64 = self - .db - .get_setting("soar_max_ttl_secs") - .ok() - .flatten() - .and_then(|v| v.parse().ok()) - .unwrap_or(86400); + // Two settings read together off the tokio worker thread — r2d2's + // pool.get() and rusqlite are blocking, so back-to-back calls inside + // an async fn can stall the executor under burst load. + let (max_ttl, max_cap) = read_block_caps(self.db.clone()).await?; if ttl_secs > max_ttl { Err(SoarError::InvalidTtl(ttl_secs, max_ttl))?; } - // Atomically check cap and reserve a slot using CAS loop (runtime-configurable via DB) - let max_cap: u32 = self - .db - .get_setting("soar_max_auto_block_cap") - .ok() - .flatten() - .and_then(|v| v.parse().ok()) - .unwrap_or(100); + // Atomically check cap and reserve a slot using CAS loop. loop { let current_count = self.active_block_count.load(Ordering::SeqCst); if current_count >= max_cap { @@ -175,13 +164,19 @@ impl SoarEngine { let expires_at = Utc::now() + ChronoDuration::seconds(ttl_secs as i64); let expires_str = expires_at.format("%Y-%m-%d %H:%M:%S").to_string(); - // tx-1 (R2 mitigation): atomically write soar_block_rules + acl_rules. - // Either both commit or both roll back — no half-state possible. + // Atomically write soar_block_rules + acl_rules. Either both commit + // or both roll back — no half-state possible. Offloaded so the + // SQLite WAL fsync can't block the tokio worker. let ip_version = ip_version_from_str(&event.source_ip); - if let Err(e) = self - .db - .commit_soar_block_to_db(&event.source_ip, ip_version, playbook_id, &expires_str) - { + let commit_result = commit_block_blocking( + self.db.clone(), + event.source_ip.clone(), + ip_version, + playbook_id, + expires_str.clone(), + ) + .await; + if let Err(e) = commit_result { // DB tx rolled back both rows; now roll back the eBPF block. if let Err(unblock_err) = self.access_control.unblock_ip(&event.source_ip) { log!(SoarLog::EventHandlingFailed(format!( @@ -189,7 +184,7 @@ impl SoarEngine { event.source_ip, unblock_err ))); // Write to pending_unblock table so recovery can retry later - if let Err(pend_err) = self.db.insert_pending_unblock(&event.source_ip) { + if let Err(pend_err) = insert_pending_unblock_blocking(self.db.clone(), event.source_ip.clone()).await { log!(SoarLog::EventHandlingFailed(format!( "CRITICAL: Failed to queue pending unblock for IP {}: {}", event.source_ip, pend_err @@ -220,13 +215,7 @@ impl SoarEngine { Err(SoarError::InvalidRateLimitFactor(factor))?; } - let max_ttl: u64 = self - .db - .get_setting("soar_max_ttl_secs") - .ok() - .flatten() - .and_then(|v| v.parse().ok()) - .unwrap_or(86400); + let max_ttl = read_max_ttl(self.db.clone()).await; if ttl_secs > max_ttl { Err(SoarError::InvalidTtl(ttl_secs, max_ttl))?; } @@ -449,3 +438,60 @@ impl SoarEngine { Ok(()) } } + +use std::sync::Arc; + +use crate::interface::port::app_repo::AppRepo; + +/// Read both block-related caps in a single offloaded blocking call so the +/// async caller pays one spawn_blocking hop instead of two. +async fn read_block_caps(db: Arc) -> Result<(u64, u32), Error> { + spawn_blocking(move || { + let max_ttl: u64 = db + .get_setting("soar_max_ttl_secs") + .ok() + .flatten() + .and_then(|v| v.parse().ok()) + .unwrap_or(86400); + let max_cap: u32 = db + .get_setting("soar_max_auto_block_cap") + .ok() + .flatten() + .and_then(|v| v.parse().ok()) + .unwrap_or(100); + Ok::<_, Error>((max_ttl, max_cap)) + }) + .await + .map_err(|e| SoarError::ActionFailed("read_block_caps", e))? +} + +async fn read_max_ttl(db: Arc) -> u64 { + spawn_blocking(move || { + db.get_setting("soar_max_ttl_secs") + .ok() + .flatten() + .and_then(|v| v.parse().ok()) + .unwrap_or(86400) + }) + .await + .unwrap_or(86400) +} + +async fn commit_block_blocking( + db: Arc, + source_ip: String, + ip_version: u8, + playbook_id: i64, + expires_str: String, +) -> Result<(), Error> { + spawn_blocking(move || db.commit_soar_block_to_db(&source_ip, ip_version, playbook_id, &expires_str)) + .await + .map_err(|e| SoarError::ActionFailed("commit_soar_block_to_db", e))??; + Ok(()) +} + +async fn insert_pending_unblock_blocking(db: Arc, source_ip: String) -> Result { + spawn_blocking(move || db.insert_pending_unblock(&source_ip)) + .await + .map_err(|e| SoarError::ActionFailed("insert_pending_unblock", e))? +} diff --git a/net-guardia/src/core/soar/rate_limit_owner.rs b/net-guardia/src/core/soar/rate_limit_owner.rs index cdb520a..c66813c 100644 --- a/net-guardia/src/core/soar/rate_limit_owner.rs +++ b/net-guardia/src/core/soar/rate_limit_owner.rs @@ -53,6 +53,11 @@ impl RateLimitOwnerHandle { /// Spawn the owner task on the current tokio runtime. The owner holds /// the only mutating references to the DB rate-limit settings and the /// eBPF rate-limit map for the duration of an Adjust / Restore batch. + /// + /// Each command body runs inside `spawn_blocking` because both halves + /// (rusqlite via r2d2 and eBPF map writes) are synchronous I/O — running + /// them directly inside the async owner loop would block the tokio + /// worker thread for the entire adjust/restore batch. pub fn spawn(db: Arc, rate_limit: Arc) -> Self { let (tx, mut rx) = mpsc::channel::(RATE_LIMIT_CMD_CHANNEL_CAPACITY); tokio::spawn(async move { @@ -65,11 +70,26 @@ impl RateLimitOwnerHandle { attack_type, reply, } => { - let result = adjust(&db, rate_limit.as_ref(), factor, ttl_secs, &source_ip, &attack_type); + let db_c = db.clone(); + let rl_c = rate_limit.clone(); + let join = tokio::task::spawn_blocking(move || { + adjust(&db_c, rl_c.as_ref(), factor, ttl_secs, &source_ip, &attack_type) + }) + .await; + let result = match join { + Ok(r) => r, + Err(e) => Err(SoarError::RateLimitOwnerJoinFailed(e.to_string()).into()), + }; let _ = reply.send(result); } RateLimitCmd::RestoreIfExpired { reply } => { - let result = restore_if_expired(&db, rate_limit.as_ref()); + let db_c = db.clone(); + let rl_c = rate_limit.clone(); + let join = tokio::task::spawn_blocking(move || restore_if_expired(&db_c, rl_c.as_ref())).await; + let result = match join { + Ok(r) => r, + Err(e) => Err(SoarError::RateLimitOwnerJoinFailed(e.to_string()).into()), + }; let _ = reply.send(result); } } diff --git a/net-guardia/src/infrastructure/audit_logger.rs b/net-guardia/src/infrastructure/audit_logger.rs index bee253e..54b0e81 100644 --- a/net-guardia/src/infrastructure/audit_logger.rs +++ b/net-guardia/src/infrastructure/audit_logger.rs @@ -29,7 +29,7 @@ impl AuditLogger { loop { match rx.recv().await { Ok(event) => { - this.handle_audit_event(&event); + this.handle_audit_event(event).await; } Err(RecvError::Lagged(n)) => { log!(AuditLog::AuditLagged(n)); @@ -52,7 +52,7 @@ impl AuditLogger { loop { match rx.recv().await { Ok(event) => { - this.handle_drift_event(&event); + this.handle_drift_event(event).await; } Err(RecvError::Lagged(n)) => { log!(AuditLog::AuditLagged(n)); @@ -69,21 +69,36 @@ impl AuditLogger { } } - fn handle_audit_event(&self, event: &AuditEvent) { + async fn handle_audit_event(&self, event: AuditEvent) { // Always emit a structured log line log!(AuditLog::AuditEvent(event.actor.clone(), event.action.clone(),)); - // Attempt DB insert; on failure, log a warning but do not panic - if let Err(e) = self.db.insert_audit_log(&event.actor, &event.action, &event.detail) { - log!(AuditLog::AuditDbWriteFailed( - e.to_string(), - event.actor.clone(), - event.action.clone(), - )); + // SQLite insert via r2d2 is blocking; offload so it can't stall the + // tokio worker that drains the broadcast channel. The actor/action + // strings are cloned for the log call above; the event itself moves + // into the blocking task. + let db = self.db.clone(); + let join = tokio::task::spawn_blocking(move || { + db.insert_audit_log(&event.actor, &event.action, &event.detail) + .map_err(|e| (e.to_string(), event.actor, event.action)) + }) + .await; + match join { + Ok(Ok(())) => {} + Ok(Err((err, actor, action))) => { + log!(AuditLog::AuditDbWriteFailed(err, actor, action)); + } + Err(join_err) => { + log!(AuditLog::AuditDbWriteFailed( + format!("blocking task join failed: {join_err}"), + "".to_string(), + "".to_string(), + )); + } } } - fn handle_drift_event(&self, event: &DriftDetectedEvent) { + async fn handle_drift_event(&self, event: DriftDetectedEvent) { let detail = serde_json::json!({ "drifted_features": event.drifted_features, "max_deviation": event.max_deviation, @@ -92,8 +107,22 @@ impl AuditLogger { log!(AuditLog::AuditDriftEvent(event.drifted_features.len())); - if let Err(e) = self.db.insert_audit_log("system", "ml_drift_detected", &detail) { - log!(AuditLog::AuditDriftDbWriteFailed(e.to_string())); + let db = self.db.clone(); + let join = tokio::task::spawn_blocking(move || { + db.insert_audit_log("system", "ml_drift_detected", &detail) + .map_err(|e| e.to_string()) + }) + .await; + match join { + Ok(Ok(())) => {} + Ok(Err(err)) => { + log!(AuditLog::AuditDriftDbWriteFailed(err)); + } + Err(join_err) => { + log!(AuditLog::AuditDriftDbWriteFailed(format!( + "blocking task join failed: {join_err}" + ))); + } } } } diff --git a/net-guardia/src/infrastructure/suricata_monitor.rs b/net-guardia/src/infrastructure/suricata_monitor.rs index c08fe41..63d7b84 100644 --- a/net-guardia/src/infrastructure/suricata_monitor.rs +++ b/net-guardia/src/infrastructure/suricata_monitor.rs @@ -115,6 +115,14 @@ impl SuricataMonitor { if raw.is_empty() { return; } + // Cheap pre-filter: a busy Suricata writes flow/stats/dns/http/fileinfo + // lines that vastly outnumber alerts. Parsing every line into + // serde_json::Value just to drop it dominates CPU on this tail. The + // substring match is a sound over-approximation — false positives go + // through the full parse + the strict event_type=="alert" check below. + if !raw.contains("\"event_type\":\"alert\"") { + return; + } let v: serde_json::Value = match serde_json::from_str(raw) { Ok(v) => v, Err(_) => return, diff --git a/net-guardia/src/model/error/soar.rs b/net-guardia/src/model/error/soar.rs index b763e9d..2403140 100644 --- a/net-guardia/src/model/error/soar.rs +++ b/net-guardia/src/model/error/soar.rs @@ -59,5 +59,9 @@ traceable! { #[no_source] #[error("Rate-limit owner task is unavailable (channel closed)")] RateLimitOwnerUnavailable => tracing::Level::ERROR, + + #[no_source] + #[error("Rate-limit owner blocking task panicked or was cancelled: {detail}")] + RateLimitOwnerJoinFailed { detail: String } => tracing::Level::ERROR, } }