diff --git a/net-guardia-frontend b/net-guardia-frontend index 0e600c5..00d347c 160000 --- a/net-guardia-frontend +++ b/net-guardia-frontend @@ -1 +1 @@ -Subproject commit 0e600c5f4956aa599beff09413011b088e848fe9 +Subproject commit 00d347c5eae3ed32f595b0a3553601f16bedfa7e diff --git a/net-guardia/src/adapter/http/logs.rs b/net-guardia/src/adapter/http/logs.rs index ec59e8d..fdce5a3 100644 --- a/net-guardia/src/adapter/http/logs.rs +++ b/net-guardia/src/adapter/http/logs.rs @@ -4,7 +4,9 @@ use std::path::Path; use std::time::UNIX_EPOCH; use actix_web::{HttpResponse, Scope, web}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; + +use crate::core::observability::log_buffer::{self, LogEntry}; /// Hardcoded log directory — not configurable via API to prevent directory traversal. const LOG_DIR: &str = "logs"; @@ -12,6 +14,15 @@ const LOG_DIR: &str = "logs"; /// Maximum downloadable log file size (50 MB). Prevents OOM from reading huge files. const MAX_DOWNLOAD_SIZE: u64 = 50 * 1024 * 1024; +/// Default page size for `/live` when the client does not specify `limit`. +/// Chosen so a 2 s poll against a DEBUG-chatty deployment catches up in +/// one round-trip without being absurd payload-wise. +const LIVE_DEFAULT_LIMIT: usize = 500; + +/// Hard cap on `/live?limit=` — prevents pathological clients from asking +/// for the entire buffer at once. +const LIVE_MAX_LIMIT: usize = 2_000; + /// Validate log filename: only alphanumeric, dots, underscores, hyphens. /// Prevents path traversal. fn is_valid_log_filename(name: &str) -> bool { @@ -25,9 +36,52 @@ fn is_valid_log_filename(name: &str) -> bool { pub fn initialize() -> Scope { web::scope("/logs") .route("", web::get().to(list_logs)) + .route("/live", web::get().to(live_logs)) .route("/{filename}", web::get().to(download_log)) } +#[derive(Deserialize)] +struct LiveQuery { + #[serde(default)] + since_id: Option, + #[serde(default)] + limit: Option, + #[serde(default)] + min_level: Option, +} + +#[derive(Serialize)] +struct LiveResponse { + entries: Vec, + next_id: u64, + total_buffered: usize, + dropped_oldest: bool, +} + +async fn live_logs(query: web::Query) -> HttpResponse { + let since_id = query.since_id.unwrap_or(0); + let limit = query.limit.unwrap_or(LIVE_DEFAULT_LIMIT).clamp(1, LIVE_MAX_LIMIT); + let min_severity = query + .min_level + .as_deref() + .map(|s| log_buffer::level_severity(&s.to_ascii_uppercase())) + .unwrap_or(log_buffer::level_severity("TRACE")); + + let snap = log_buffer::snapshot(since_id, min_severity, limit); + // Signal to the UI that it lagged enough for the ring to evict rows + // between polls. Frontend can warn "older entries dropped" without + // silently skipping a gap. + let dropped_oldest = since_id > 0 && snap.entries.first().is_some_and(|e| e.id > since_id + 1); + let next_id = snap.entries.last().map(|e| e.id).unwrap_or(snap.latest_id); + + HttpResponse::Ok().json(LiveResponse { + entries: snap.entries, + next_id, + total_buffered: snap.total, + dropped_oldest, + }) +} + #[derive(Serialize)] struct LogFileEntry { name: String, diff --git a/net-guardia/src/adapter/persistence/repository.rs b/net-guardia/src/adapter/persistence/repository.rs index 839a5b7..28522fb 100644 --- a/net-guardia/src/adapter/persistence/repository.rs +++ b/net-guardia/src/adapter/persistence/repository.rs @@ -47,7 +47,25 @@ impl r2d2::CustomizeConnection for Sqlite // Use a parameterised query to avoid SQL-injection via the key value. conn.pragma_update(None, "key", key)?; } - conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?; + // PRAGMA tuning notes: + // - `journal_mode=WAL`: many concurrent readers + one writer; the only + // journal mode that survives crashes without losing committed rows. + // - `synchronous=NORMAL`: canonical pairing with WAL — `FULL` adds an + // extra fsync per commit that buys no durability guarantees beyond + // what WAL already provides for a power-loss event. + // - `busy_timeout=5000`: WAL still serializes writers (SOAR, audit, + // drift, SQL hooks all share one DB), and the default 0ms returns + // SQLITE_BUSY immediately on any contention. 5s gives the loser + // enough time to wait out a normal commit (sub-ms) without masking + // genuine deadlocks. + // - `foreign_keys=ON`: enforce FK constraints at the connection + // level (SQLite's default is OFF for backwards compatibility). + conn.execute_batch( + "PRAGMA journal_mode=WAL; \ + PRAGMA synchronous=NORMAL; \ + PRAGMA busy_timeout=5000; \ + PRAGMA foreign_keys=ON;", + )?; Ok(()) } } diff --git a/net-guardia/src/adapter/websocket/fusion_websocket.rs b/net-guardia/src/adapter/websocket/fusion_websocket.rs new file mode 100644 index 0000000..3ead7c7 --- /dev/null +++ b/net-guardia/src/adapter/websocket/fusion_websocket.rs @@ -0,0 +1,192 @@ +//! WebSocket bridge for post-fusion threat events. +//! +//! `/ws/fusion` subscribes to the `ThreatDetectedEvent` broadcast that the +//! `DetectionOrchestrator` already publishes through `CommunicationManager` +//! (the same stream SOAR consumes). Each event is wrapped with a server-side +//! `ts` (unix seconds) so the dashboard can render relative timestamps +//! without doing the conversion itself. +//! +//! Distinct from `/ws/alerts` (flow-level ML detections via `MLAlert`): +//! this stream is the **fused, per-IP, multi-source** view that drives the +//! Overview "Recent Threats" card and the sources-agreed chip. Treating +//! them as one channel would conflate two bounded contexts — see +//! `docs/strategy/DOMAIN_MAP.md` for the BC split rationale. + +use std::time::{SystemTime, UNIX_EPOCH}; + +use actix_web::rt::spawn; +use actix_web::{HttpRequest, HttpResponse, Result, web}; +use actix_ws::{Message, MessageStream, Session, handle}; +use futures_util::StreamExt; +use macros::log; +use tokio::sync::broadcast; +use tokio::sync::broadcast::error::RecvError; + +use crate::infrastructure::communication_manager::CommunicationManager; +use crate::model::error::http::HttpError; +use crate::model::error::misc::MiscError; +use crate::model::event::ThreatDetectedEvent; +use crate::model::log::http::HttpLog; + +pub async fn websocket_fusion( + req: HttpRequest, + body: web::Payload, + comm: web::Data, +) -> Result { + let (response, session, msg_stream) = handle(&req, body)?; + + let broadcast_rx = match comm.subscribe_event::() { + Ok(rx) => rx, + Err(e) => { + log!(HttpLog::FusionSubscribeFailed(e.to_string())); + return Ok(HttpResponse::InternalServerError().json(serde_json::json!({ + "error": "fusion event channel not registered", + }))); + } + }; + + spawn(async move { + handle_fusion_connection(session, msg_stream, broadcast_rx).await; + }); + + Ok(response) +} + +async fn handle_fusion_connection( + mut session: Session, + mut msg_stream: MessageStream, + mut broadcast_rx: broadcast::Receiver, +) { + loop { + tokio::select! { + msg_result = msg_stream.next() => { + if !handle_client_message(&mut session, msg_result).await { + break; + } + }, + broadcast_result = broadcast_rx.recv() => { + match broadcast_result { + Ok(event) => { + if !send_event(&mut session, &event).await { + break; + } + } + Err(RecvError::Lagged(skipped)) => { + log!(HttpLog::WebSocketLagged(skipped)); + continue; + } + Err(RecvError::Closed) => { + break; + } + } + }, + } + } + + let _ = session.close(None).await; +} + +async fn handle_client_message( + session: &mut Session, + msg_result: Option>, +) -> bool { + match msg_result { + Some(Ok(Message::Text(_))) => true, + Some(Ok(Message::Ping(bytes))) => session.pong(&bytes).await.is_ok(), + Some(Ok(Message::Close(reason))) => { + let _ = (session.clone()).close(reason).await; + false + } + Some(Err(err)) => { + log!(HttpError::WebSocketError(err)); + false + } + None => false, + _ => true, + } +} + +/// Wrap each event in `{ts, ...event_fields}`. The `ts` is a server-stamped +/// unix-seconds value so the client can render "5s ago" without inferring +/// the time from the audit chain. All declared fields of +/// `ThreatDetectedEvent` flow through verbatim via the event's own +/// `Serialize` derive — no field whitelist to drift out of date. +async fn send_event(session: &mut Session, event: &ThreatDetectedEvent) -> bool { + let ts = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + + let payload = match serde_json::to_value(event) { + Ok(serde_json::Value::Object(mut map)) => { + map.insert("ts".to_string(), serde_json::Value::from(ts)); + serde_json::Value::Object(map) + } + // The derived Serialize on a struct always produces an Object — + // this branch only fires if the type changes shape in a future + // refactor. Falling back to the raw value keeps the stream alive. + Ok(other) => other, + Err(err) => { + log!(MiscError::SerializeError(err)); + return false; + } + }; + + match serde_json::to_string(&payload) { + Ok(json) => session.text(json).await.is_ok(), + Err(err) => { + log!(MiscError::SerializeError(err)); + false + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::event::DetectionSource; + + fn sample_event() -> ThreatDetectedEvent { + ThreatDetectedEvent { + attack_type: "brute_force".to_string(), + confidence: 0.92, + source_ip: "203.0.113.10".to_string(), + dest_ip: "10.0.0.1".to_string(), + flow_count: 3, + packet_rate: 12.5, + protocol: 6, + geoip_country: Some("CN".to_string()), + is_repeat_offender: true, + sources: vec![DetectionSource::ML, DetectionSource::Suricata], + active_source_count: 2, + fused_confidence: 0.99, + ae_score: 0.0, + anomaly_score: 0.0, + c2_score: 0.0, + } + } + + #[test] + fn event_serializes_with_canonical_source_strings() { + let event = sample_event(); + let json = serde_json::to_value(&event).expect("serialize event"); + let sources = json["sources"].as_array().expect("sources array"); + assert_eq!(sources[0], "ML"); + assert_eq!(sources[1], "Suricata"); + assert_eq!(json["active_source_count"], 2); + assert_eq!(json["geoip_country"], "CN"); + } + + #[test] + fn envelope_adds_ts_field_to_event_object() { + // The `send_event` wire path inserts `ts` into the event's own + // serde object; mirror that here without an actix session so the + // wrapping logic stays covered when the orchestrator schema evolves. + let event = sample_event(); + let mut value = serde_json::to_value(&event).expect("serialize event"); + let object = value.as_object_mut().expect("expected object shape"); + object.insert("ts".to_string(), serde_json::Value::from(1_700_000_000_u64)); + assert_eq!(value["ts"], 1_700_000_000_u64); + assert_eq!(value["attack_type"], "brute_force"); + } +} diff --git a/net-guardia/src/adapter/websocket/mod.rs b/net-guardia/src/adapter/websocket/mod.rs index a488f84..d435fd2 100644 --- a/net-guardia/src/adapter/websocket/mod.rs +++ b/net-guardia/src/adapter/websocket/mod.rs @@ -1,5 +1,6 @@ pub mod alert_websocket; pub mod drop_websocket; pub mod flow_websocket; +pub mod fusion_websocket; pub mod health_websocket; pub mod routes; diff --git a/net-guardia/src/adapter/websocket/routes.rs b/net-guardia/src/adapter/websocket/routes.rs index 1bf7f0a..a97ab45 100644 --- a/net-guardia/src/adapter/websocket/routes.rs +++ b/net-guardia/src/adapter/websocket/routes.rs @@ -1,10 +1,11 @@ use actix_web::{HttpRequest, HttpResponse, Responder, Scope, web}; use serde::Deserialize; -use super::{alert_websocket, drop_websocket, flow_websocket, health_websocket}; +use super::{alert_websocket, drop_websocket, flow_websocket, fusion_websocket, health_websocket}; use crate::adapter::ebpf::drop_monitor::DropMonitor; use crate::core::auth::jwt::JwtService; use crate::core::ml::alert::MLAlert; +use crate::infrastructure::communication_manager::CommunicationManager; use crate::infrastructure::health::SystemHealth; use crate::infrastructure::statistics::FlowStatistics; @@ -17,6 +18,7 @@ pub fn initialize() -> Scope { web::scope("/ws") .route("/health", web::get().to(health_ws)) .route("/alerts", web::get().to(alerts_ws)) + .route("/fusion", web::get().to(fusion_ws)) .route("/flows", web::get().to(flows_ws)) .route("/drops", web::get().to(drops_ws)) } @@ -84,6 +86,24 @@ async fn alerts_ws( } } +async fn fusion_ws( + req: HttpRequest, + stream: web::Payload, + comm: web::Data, + query: web::Query, + jwt: web::Data, +) -> impl Responder { + if let Err(resp) = validate_ws_token(&req, &query, &jwt) { + return resp; + } + match fusion_websocket::websocket_fusion(req, stream, comm).await { + Ok(response) => response, + Err(err) => { + HttpResponse::InternalServerError().json(serde_json::json!({"error": format!("WebSocket error: {}", err)})) + } + } +} + async fn flows_ws( req: HttpRequest, stream: web::Payload, diff --git a/net-guardia/src/core/mod.rs b/net-guardia/src/core/mod.rs index 334ac08..4ed84f9 100644 --- a/net-guardia/src/core/mod.rs +++ b/net-guardia/src/core/mod.rs @@ -7,6 +7,7 @@ pub mod dns_filter_service; pub mod email; pub mod ml; pub mod notification_service; +pub mod observability; pub mod playbook_service; pub mod rate_limit_service; pub mod report; diff --git a/net-guardia/src/core/observability/log_buffer.rs b/net-guardia/src/core/observability/log_buffer.rs new file mode 100644 index 0000000..4eaef81 --- /dev/null +++ b/net-guardia/src/core/observability/log_buffer.rs @@ -0,0 +1,304 @@ +use std::collections::VecDeque; +use std::fmt::{Arguments, Debug, Write as _}; +use std::sync::OnceLock; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use parking_lot::Mutex; +use serde::Serialize; +use tracing::field::{Field, Visit}; +use tracing::{Event, Level, Subscriber}; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::Context; + +/// Ring-buffer capacity. Tuned for ~30 min of INFO traffic on a small SOC +/// deployment; DEBUG floods will churn faster. +const DEFAULT_CAPACITY: usize = 5_000; + +/// Per-entry payload cap. Guards against pathological debug logs from +/// bursting the buffer. +const MAX_MESSAGE_BYTES: usize = 8_192; + +/// Monotonic id allocator. Clients use `since_id` to resume tailing. +/// u64 never wraps in practice (2^64 events at 1 µs/event ≈ 584 000 years). +static NEXT_ID: AtomicU64 = AtomicU64::new(1); + +static BUFFER: OnceLock = OnceLock::new(); + +#[derive(Clone, Debug, Serialize)] +pub struct LogEntry { + pub id: u64, + pub ts_ms: u64, + pub level: &'static str, + pub target: String, + pub message: String, +} + +struct LogRingBuffer { + entries: Mutex>, + capacity: usize, +} + +impl LogRingBuffer { + fn new(capacity: usize) -> Self { + Self { + entries: Mutex::new(VecDeque::with_capacity(capacity)), + capacity, + } + } + + fn push(&self, entry: LogEntry) { + let mut guard = self.entries.lock(); + if guard.len() >= self.capacity { + guard.pop_front(); + } + guard.push_back(entry); + } + + fn snapshot(&self, since_id: u64, min_severity: u8, limit: usize) -> Snapshot { + let guard = self.entries.lock(); + let total = guard.len(); + let latest_id = guard.back().map(|e| e.id).unwrap_or(0); + let entries: Vec = guard + .iter() + .filter(|e| e.id > since_id && level_severity(e.level) <= min_severity) + .take(limit) + .cloned() + .collect(); + Snapshot { + entries, + latest_id, + total, + } + } +} + +pub struct Snapshot { + pub entries: Vec, + pub latest_id: u64, + pub total: usize, +} + +/// Get a snapshot for the `/api/logs/live` endpoint. +/// +/// `min_severity` follows tracing level numeric ordering (ERROR=1…TRACE=5); +/// an entry at level L is included when `level_severity(L) <= min_severity`. +/// Returns an empty snapshot when the buffer has not been installed yet +/// (tests, dry-runs). +pub fn snapshot(since_id: u64, min_severity: u8, limit: usize) -> Snapshot { + match BUFFER.get() { + Some(buf) => buf.snapshot(since_id, min_severity, limit), + None => Snapshot { + entries: Vec::new(), + latest_id: 0, + total: 0, + }, + } +} + +/// Map a level string to severity rank. Unknown strings sort as TRACE so +/// they are only visible when the caller asks for everything. +pub fn level_severity(level: &str) -> u8 { + match level { + "ERROR" => 1, + "WARN" => 2, + "INFO" => 3, + "DEBUG" => 4, + _ => 5, + } +} + +fn level_str(level: &Level) -> &'static str { + match *level { + Level::ERROR => "ERROR", + Level::WARN => "WARN", + Level::INFO => "INFO", + Level::DEBUG => "DEBUG", + Level::TRACE => "TRACE", + } +} + +fn now_unix_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// [`Layer`] that appends each formatted event into the in-memory ring +/// buffer so the UI can tail logs without round-tripping the filesystem. +pub struct LogBufferLayer; + +impl LogBufferLayer { + pub fn new() -> Self { + let _ = BUFFER.set(LogRingBuffer::new(DEFAULT_CAPACITY)); + Self + } +} + +impl Default for LogBufferLayer { + fn default() -> Self { + Self::new() + } +} + +impl Layer for LogBufferLayer { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + let Some(buf) = BUFFER.get() else { + return; + }; + let metadata = event.metadata(); + let mut visitor = MessageVisitor::default(); + event.record(&mut visitor); + let mut message = visitor.into_message(); + if message.len() > MAX_MESSAGE_BYTES { + message.truncate(MAX_MESSAGE_BYTES); + message.push_str("…[truncated]"); + } + let entry = LogEntry { + id: NEXT_ID.fetch_add(1, Ordering::Relaxed), + ts_ms: now_unix_ms(), + level: level_str(metadata.level()), + target: metadata.target().to_string(), + message, + }; + buf.push(entry); + } +} + +/// Collects `message` plus remaining fields as `key=value` pairs. `tracing` +/// macros emit the format-args body under the `message` field; structured +/// fields come through [`Visit::record_*`] for the respective primitive. +#[derive(Default)] +struct MessageVisitor { + message: String, + extra: String, +} + +impl MessageVisitor { + fn into_message(mut self) -> String { + if self.extra.is_empty() { + self.message + } else if self.message.is_empty() { + self.extra + } else { + self.message.push(' '); + self.message.push_str(&self.extra); + self.message + } + } + + fn push_extra(&mut self, name: &str, value: Arguments<'_>) { + if !self.extra.is_empty() { + self.extra.push(' '); + } + let _ = write!(self.extra, "{}={}", name, value); + } +} + +impl Visit for MessageVisitor { + fn record_debug(&mut self, field: &Field, value: &dyn Debug) { + if field.name() == "message" { + let _ = write!(self.message, "{:?}", value); + } else { + self.push_extra(field.name(), format_args!("{:?}", value)); + } + } + + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() == "message" { + self.message.push_str(value); + } else { + self.push_extra(field.name(), format_args!("{}", value)); + } + } + + fn record_i64(&mut self, field: &Field, value: i64) { + self.push_extra(field.name(), format_args!("{}", value)); + } + + fn record_u64(&mut self, field: &Field, value: u64) { + self.push_extra(field.name(), format_args!("{}", value)); + } + + fn record_f64(&mut self, field: &Field, value: f64) { + self.push_extra(field.name(), format_args!("{}", value)); + } + + fn record_bool(&mut self, field: &Field, value: bool) { + self.push_extra(field.name(), format_args!("{}", value)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_entry(id: u64, level: &'static str, message: &str) -> LogEntry { + LogEntry { + id, + ts_ms: 0, + level, + target: "test".into(), + message: message.into(), + } + } + + #[test] + fn severity_ordering() { + assert!(level_severity("ERROR") < level_severity("WARN")); + assert!(level_severity("WARN") < level_severity("INFO")); + assert!(level_severity("INFO") < level_severity("DEBUG")); + assert!(level_severity("DEBUG") < level_severity("TRACE")); + assert_eq!(level_severity("unknown"), 5); + } + + #[test] + fn ring_buffer_drops_oldest_at_capacity() { + let buf = LogRingBuffer::new(3); + for id in 1..=5 { + buf.push(make_entry(id, "INFO", "m")); + } + let snap = buf.snapshot(0, level_severity("TRACE"), 100); + let ids: Vec = snap.entries.iter().map(|e| e.id).collect(); + assert_eq!(ids, vec![3, 4, 5]); + assert_eq!(snap.latest_id, 5); + assert_eq!(snap.total, 3); + } + + #[test] + fn snapshot_filters_since_id_and_severity() { + let buf = LogRingBuffer::new(16); + buf.push(make_entry(1, "INFO", "first")); + buf.push(make_entry(2, "DEBUG", "noisy")); + buf.push(make_entry(3, "ERROR", "boom")); + + let snap = buf.snapshot(1, level_severity("INFO"), 100); + let levels: Vec<&str> = snap.entries.iter().map(|e| e.level).collect(); + assert_eq!(levels, vec!["ERROR"]); + assert_eq!(snap.latest_id, 3); + } + + #[test] + fn snapshot_respects_limit() { + let buf = LogRingBuffer::new(16); + for id in 1..=10 { + buf.push(make_entry(id, "INFO", "m")); + } + let snap = buf.snapshot(0, level_severity("TRACE"), 4); + assert_eq!(snap.entries.len(), 4); + assert_eq!(snap.latest_id, 10); + } + + #[test] + fn visitor_concatenates_message_and_structured_fields() { + let mut v = MessageVisitor::default(); + v.push_extra("count", format_args!("{}", 42u64)); + v.push_extra("ok", format_args!("{}", true)); + v.message.push_str("hello"); + let out = v.into_message(); + assert!(out.contains("hello")); + assert!(out.contains("count=42")); + assert!(out.contains("ok=true")); + } +} diff --git a/net-guardia/src/core/observability/mod.rs b/net-guardia/src/core/observability/mod.rs new file mode 100644 index 0000000..bb1f13a --- /dev/null +++ b/net-guardia/src/core/observability/mod.rs @@ -0,0 +1 @@ +pub mod log_buffer; diff --git a/net-guardia/src/model/event.rs b/net-guardia/src/model/event.rs index ce67f8f..7cf48f4 100644 --- a/net-guardia/src/model/event.rs +++ b/net-guardia/src/model/event.rs @@ -1,13 +1,20 @@ use std::fmt; use std::str::FromStr; +use serde::Serialize; + use crate::interface::communication::event::Event; // -- Detection Source --------------------------------------------------------- /// Identifies which detection subsystem produced a detection. /// Used for attribution tracking and cross-source deduplication. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +/// +/// Serialized as the canonical `Display` form ("ML", "Suricata", "Beaconing", +/// "Correlation") so the WebSocket wire matches the SOAR `SingleSourceHigh` +/// `value` field — frontend rendering and playbook authoring share one +/// vocabulary. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)] pub enum DetectionSource { ML, Correlation, @@ -66,8 +73,11 @@ pub struct DetectionEvent { // -- Threat Events ------------------------------------------------------------ /// Fired when the DetectionOrchestrator emits a deduplicated, enriched threat. -/// Consumed by the SOAR engine to trigger automated responses. -#[derive(Debug, Clone)] +/// Consumed by the SOAR engine to trigger automated responses, and broadcast +/// to the dashboard over `/ws/fusion` so the operator's "Recent Threats" +/// stream surfaces post-fusion (multi-source) detections rather than raw +/// per-flow ML alerts. +#[derive(Debug, Clone, Serialize)] pub struct ThreatDetectedEvent { pub attack_type: String, pub confidence: f32, diff --git a/net-guardia/src/model/log/http.rs b/net-guardia/src/model/log/http.rs index 9f27e9f..2809636 100644 --- a/net-guardia/src/model/log/http.rs +++ b/net-guardia/src/model/log/http.rs @@ -11,5 +11,8 @@ loggable! { #[error("Setup HTTP server error: {error}")] SetupServerError { error: String } => tracing::Level::ERROR, + + #[error("Fusion WebSocket failed to subscribe to ThreatDetectedEvent: {err}")] + FusionSubscribeFailed { err: String } => tracing::Level::ERROR, } } diff --git a/net-guardia/src/utils/logging.rs b/net-guardia/src/utils/logging.rs index 73cdd33..5828e6f 100644 --- a/net-guardia/src/utils/logging.rs +++ b/net-guardia/src/utils/logging.rs @@ -10,6 +10,7 @@ use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::reload; use tracing_subscriber::util::SubscriberInitExt; +use crate::core::observability::log_buffer::LogBufferLayer; use crate::model::error::Error; use crate::model::error::io::IOError; @@ -78,6 +79,7 @@ impl Logging { .with(filter_layer) .with(stdout_layer) .with(file_layer) + .with(LogBufferLayer::new()) .init(); // Store type-erased handle for runtime log level changes @@ -102,14 +104,47 @@ impl Logging { handle.reload_filter(new_filter)?; - Ok(parsed_level.to_string()) + Ok(parsed_level.to_string().to_lowercase()) } - /// Get the current log level filter string. + /// Get the current global log level as a bare lowercase directive — + /// e.g. `"info"`, not the full `"maxminddb=warn,info"` EnvFilter string. + /// Per-target overrides (like `maxminddb=warn`) are internal tuning and + /// would break the frontend `