diff --git a/macros/src/config.rs b/macros/src/config.rs index b91fdd4..e86268b 100644 --- a/macros/src/config.rs +++ b/macros/src/config.rs @@ -26,7 +26,9 @@ impl Parse for StructAttr { input.parse::()?; } } - Ok(Self { default_section: section }) + Ok(Self { + default_section: section, + }) } } @@ -71,10 +73,7 @@ struct MappedParent { // ── Parsing ──────────────────────────────────────────────────────── -fn parse_struct_mapped_settings( - input: &mut ItemStruct, - default_section: &Option, -) -> Vec { +fn parse_struct_mapped_settings(input: &mut ItemStruct, default_section: &Option) -> Vec { let mut mapped = Vec::new(); input.attrs.retain(|attr| { if !attr.path().is_ident("setting") { @@ -131,14 +130,8 @@ fn parse_struct_mapped_settings( mapped } -fn parse_field( - field: &mut syn::Field, - default_section: &Option, -) -> Option { - let idx = field - .attrs - .iter() - .position(|a| a.path().is_ident("setting"))?; +fn parse_field(field: &mut syn::Field, default_section: &Option) -> Option { + let idx = field.attrs.iter().position(|a| a.path().is_ident("setting"))?; let attr = field.attrs.remove(idx); let mut is_flatten = false; @@ -434,15 +427,11 @@ pub fn config_settings_impl(attr: TokenStream, item: TokenStream) -> TokenStream let struct_attr = syn::parse_macro_input!(attr as StructAttr); let mut input = syn::parse_macro_input!(item as ItemStruct); - let mapped_settings = - parse_struct_mapped_settings(&mut input, &struct_attr.default_section); + let mapped_settings = parse_struct_mapped_settings(&mut input, &struct_attr.default_section); let mut mapped_groups: BTreeMap> = BTreeMap::new(); for ms in mapped_settings { - mapped_groups - .entry(ms.parent.clone()) - .or_default() - .push(ms); + mapped_groups.entry(ms.parent.clone()).or_default().push(ms); } let fields = match &mut input.fields { @@ -452,11 +441,7 @@ pub fn config_settings_impl(attr: TokenStream, item: TokenStream) -> TokenStream let mut config_fields = Vec::new(); for field in &mut fields.named { - let field_name = field - .ident - .as_ref() - .expect("named field") - .to_string(); + let field_name = field.ident.as_ref().expect("named field").to_string(); if let Some(settings) = mapped_groups.remove(&field_name) { config_fields.push(ConfigField::MappedParent(MappedParent { diff --git a/net-guardia/src/adapter/http/ml.rs b/net-guardia/src/adapter/http/ml.rs index 7e83593..2ae1334 100644 --- a/net-guardia/src/adapter/http/ml.rs +++ b/net-guardia/src/adapter/http/ml.rs @@ -1,4 +1,5 @@ use actix_web::{HttpResponse, Responder, Scope, web}; +use tokio::sync::broadcast; use crate::core::identity::extractor::AuthClaims; use crate::core::inference::engine::Engine; @@ -6,7 +7,6 @@ use crate::core::inference::inference::Inference; use crate::domain::common::config::constants::AUDIT_ACTOR_SECURITY_ADMIN_PREFIX; use crate::domain::common::event::AuditEvent; use crate::domain::detection::model_adapter::ModelSourceState; -use crate::infrastructure::communication_manager::CommunicationManager; /// Permission required to forcibly revert the active ML source to dormant. /// Mirrors the upload handler's gate so swap-out and revert are symmetric: @@ -65,7 +65,7 @@ async fn get_current_model(inference: web::Data) -> impl Responder { /// the upload path's `model_swap` so both swap-in and revert are auditable. async fn delete_current_model( inference: web::Data, - comm: web::Data, + audit_tx: web::Data>, claims: AuthClaims, ) -> impl Responder { if !claims.permissions.iter().any(|p| p == DORMANT_REQUIRED_PERMISSION) { @@ -87,13 +87,11 @@ async fn delete_current_model( "before": serde_json::to_value(&before_status).unwrap_or(serde_json::Value::Null), }) .to_string(); - let _ = comm - .publish_event(AuditEvent { - actor: format!("{AUDIT_ACTOR_SECURITY_ADMIN_PREFIX}@{}", claims.username), - action: AUDIT_ACTION_MODEL_DORMANT.to_string(), - detail: audit_detail, - }) - .await; + let _ = audit_tx.send(AuditEvent { + actor: format!("{AUDIT_ACTOR_SECURITY_ADMIN_PREFIX}@{}", claims.username), + action: AUDIT_ACTION_MODEL_DORMANT.to_string(), + detail: audit_detail, + }); HttpResponse::Ok().json(serde_json::json!({ "already_dormant": false, diff --git a/net-guardia/src/adapter/http/model_upload.rs b/net-guardia/src/adapter/http/model_upload.rs index b2081ec..6053697 100644 --- a/net-guardia/src/adapter/http/model_upload.rs +++ b/net-guardia/src/adapter/http/model_upload.rs @@ -31,6 +31,7 @@ use serde_json::Value as JsonValue; use sha2::{Digest, Sha256}; use tokio::fs; use tokio::io::AsyncWriteExt; +use tokio::sync::broadcast; use tokio::task; use uuid::Uuid; @@ -44,7 +45,6 @@ use crate::domain::common::config::constants::{ use crate::domain::common::event::AuditEvent; use crate::domain::detection::manifest::{AdapterKind, ModelManifest}; use crate::domain::detection::ml_inference_config::MLInferenceConfig; -use crate::infrastructure::communication_manager::CommunicationManager; /// Multipart field names the client must use. Stable wire contract — /// the frontend form generator depends on these exact strings. @@ -133,7 +133,7 @@ pub fn initialize() -> Scope { async fn upload( app_config: web::Data>, inference: web::Data, - comm: web::Data, + audit_tx: web::Data>, promote_lock: web::Data, claims: AuthClaims, payload: Multipart, @@ -168,7 +168,7 @@ async fn upload( &staging_dir, &summary, inference.get_ref(), - comm.get_ref(), + audit_tx.get_ref(), promote_lock.get_ref(), &claims.username, batch_size, @@ -516,7 +516,7 @@ async fn validate_and_promote( staging_dir: &Path, summary: &UploadSummary, inference: &Inference, - comm: &CommunicationManager, + audit_tx: &broadcast::Sender, promote_lock: &PromoteGate, actor_username: &str, batch_size: usize, @@ -605,13 +605,11 @@ async fn validate_and_promote( "before": serde_json::to_value(&before_status).unwrap_or(JsonValue::Null), }) .to_string(); - let _ = comm - .publish_event(AuditEvent { - actor: format!("{AUDIT_ACTOR_SECURITY_ADMIN_PREFIX}@{actor_username}"), - action: AUDIT_ACTION_MODEL_SWAP.to_string(), - detail: audit_detail, - }) - .await; + let _ = audit_tx.send(AuditEvent { + actor: format!("{AUDIT_ACTOR_SECURITY_ADMIN_PREFIX}@{actor_username}"), + action: AUDIT_ACTION_MODEL_SWAP.to_string(), + detail: audit_detail, + }); Ok(PromoteReport { manifest_name: manifest.name, diff --git a/net-guardia/src/adapter/http/system.rs b/net-guardia/src/adapter/http/system.rs index 64c5878..407300c 100644 --- a/net-guardia/src/adapter/http/system.rs +++ b/net-guardia/src/adapter/http/system.rs @@ -3,11 +3,9 @@ use serde::Deserialize; use crate::core::common::config_service::ConfigService; use crate::core::identity::extractor::AuthClaims; -use crate::infrastructure::communication_manager::CommunicationManager; +use crate::infrastructure::enforce_mode_handler::EnforceModeHandler; use crate::infrastructure::logger::Logger; use crate::infrastructure::system::{ShutdownHandle, ShutdownMode}; -use crate::interface::communication::command_types::ChangeEnforceModeCommand; -use crate::interface::communication::query_types::GetEnforceModeQuery; use crate::interface::port::app_repo::AppRepo; use crate::utils::boot_time; @@ -36,8 +34,8 @@ async fn get_boot_time() -> impl Responder { HttpResponse::Ok().json(boot_time::boot_time()) } -async fn get_enforce_mode(comm: web::Data) -> impl Responder { - match comm.send_query(GetEnforceModeQuery).await { +async fn get_enforce_mode(handler: web::Data) -> impl Responder { + match handler.get_mode() { Ok(mode) => HttpResponse::Ok().json(serde_json::json!({"mode": mode})), Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({"error": e.to_string()})), } @@ -45,7 +43,7 @@ async fn get_enforce_mode(comm: web::Data) -> impl Respond async fn set_enforce_mode( body: web::Json, - comm: web::Data, + handler: web::Data, ) -> impl Responder { let mode = &body.mode; if mode != "monitor" && mode != "ml_only" && mode != "enforce" { @@ -53,7 +51,7 @@ async fn set_enforce_mode( .json(serde_json::json!({"error": "Mode must be 'monitor', 'ml_only', or 'enforce'"})); } - match comm.send_command(ChangeEnforceModeCommand { mode: mode.clone() }).await { + match handler.change_mode(mode.clone()) { Ok(_) => HttpResponse::Ok().json(serde_json::json!({"mode": mode})), Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({"error": e.to_string()})), } diff --git a/net-guardia/src/adapter/websocket/fusion_websocket.rs b/net-guardia/src/adapter/websocket/fusion_websocket.rs index f4fa700..4098ee7 100644 --- a/net-guardia/src/adapter/websocket/fusion_websocket.rs +++ b/net-guardia/src/adapter/websocket/fusion_websocket.rs @@ -1,8 +1,8 @@ //! 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 +//! `/ws/fusion` subscribes to the `ThreatDetectedEvent` broadcast channel +//! that the `DetectionOrchestrator` publishes to (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. //! @@ -26,24 +26,15 @@ use crate::domain::common::error::http::HttpError; use crate::domain::common::error::misc::MiscError; use crate::domain::common::event::ThreatDetectedEvent; use crate::domain::common::log::http::HttpLog; -use crate::infrastructure::communication_manager::CommunicationManager; pub async fn websocket_fusion( req: HttpRequest, body: web::Payload, - comm: web::Data, + threat_tx: 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", - }))); - } - }; + let broadcast_rx = threat_tx.subscribe(); spawn(async move { handle_fusion_connection(session, msg_stream, broadcast_rx).await; diff --git a/net-guardia/src/adapter/websocket/routes.rs b/net-guardia/src/adapter/websocket/routes.rs index d9acb36..be587fc 100644 --- a/net-guardia/src/adapter/websocket/routes.rs +++ b/net-guardia/src/adapter/websocket/routes.rs @@ -1,11 +1,12 @@ use actix_web::{HttpRequest, HttpResponse, Responder, Scope, web}; use serde::Deserialize; +use tokio::sync::broadcast; use super::{alert_websocket, drop_websocket, flow_websocket, fusion_websocket, health_websocket}; use crate::adapter::ebpf::drop_monitor::DropMonitor; use crate::core::identity::jwt::JwtService; use crate::core::inference::alert::MLAlert; -use crate::infrastructure::communication_manager::CommunicationManager; +use crate::domain::common::event::ThreatDetectedEvent; use crate::infrastructure::health::SystemHealth; use crate::infrastructure::statistics::FlowStatistics; @@ -86,14 +87,14 @@ async fn alerts_ws( async fn fusion_ws( req: HttpRequest, stream: web::Payload, - comm: web::Data, + threat_tx: 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 { + match fusion_websocket::websocket_fusion(req, stream, threat_tx).await { Ok(response) => response, Err(err) => { HttpResponse::InternalServerError().json(serde_json::json!({"error": format!("WebSocket error: {}", err)})) diff --git a/net-guardia/src/core/detection/orchestrator.rs b/net-guardia/src/core/detection/orchestrator.rs index 8d0c2fe..2102a91 100644 --- a/net-guardia/src/core/detection/orchestrator.rs +++ b/net-guardia/src/core/detection/orchestrator.rs @@ -5,18 +5,17 @@ use std::time::{Duration, Instant}; use arc_swap::ArcSwap; use lru::LruCache; use macros::log; +use tokio::sync::broadcast; use tokio::sync::mpsc; use tokio::time::interval; use crate::domain::common::config::AppConfig; use crate::domain::common::config::constants::{FUSION_AUDIT_ACTION, FUSION_AUDIT_ACTOR}; -use crate::domain::common::error::system::SystemError; use crate::domain::common::event::{AuditEvent, DetectionEvent, DetectionSource, ThreatDetectedEvent}; use crate::domain::detection::attack_type::translate; use crate::domain::detection::fusion_math::{FusionWindowLengths, fused_confidence}; use crate::domain::detection::log::DetectionLog; use crate::domain::detection::metrics::FusionMetrics; -use crate::infrastructure::communication_manager::CommunicationManager; use crate::interface::port::geo_lookup::GeoLookup; /// Per-source record within an in-flight dedup entry. Keeps the strongest @@ -50,7 +49,8 @@ struct DedupEntry { /// re-emit with the combined confidence `1 − ∏(1 − c_i)`. pub struct DetectionOrchestrator { rx: mpsc::Receiver, - comm: Arc, + threat_tx: broadcast::Sender, + audit_tx: broadcast::Sender, geoip: Option>, metrics: Arc, // Enrichment state @@ -70,7 +70,8 @@ impl DetectionOrchestrator { pub fn new( app_config: &Arc>, rx: mpsc::Receiver, - comm: Arc, + threat_tx: broadcast::Sender, + audit_tx: broadcast::Sender, geoip: Option>, metrics: Arc, ) -> Self { @@ -79,7 +80,8 @@ impl DetectionOrchestrator { let max_dedup = NonZero::new(fusion.max_dedup_entries.max(1)).unwrap_or(NonZero::::MIN); Self { rx, - comm, + threat_tx, + audit_tx, geoip, metrics, // SAFETY: NonZero::new on non-zero literals. @@ -251,9 +253,7 @@ impl DetectionOrchestrator { self.publish_fusion_audit(trigger_event, fused, &per_source_samples) .await; - if let Err(e) = self.comm.publish_event(threat_event).await { - log!(SystemError::MlSoarBridgeFailed(e)); - } + let _ = self.threat_tx.send(threat_event); } /// Emit a WORM AuditEvent so the eventual "why was this IP blocked?" @@ -267,9 +267,7 @@ impl DetectionOrchestrator { detail: build_fusion_audit_detail(&trigger_event.source_ip, &trigger_event.attack_type, fused, per_source), }; - if let Err(e) = self.comm.publish_event(audit).await { - log!(DetectionLog::FusionAuditPublishFailed(e.to_string())); - } + let _ = self.audit_tx.send(audit); } async fn enrich(&mut self, event: &DetectionEvent) -> ThreatDetectedEvent { @@ -358,7 +356,7 @@ impl DetectionOrchestrator { /// Serialize the WORM audit evidence payload for a fused threat emission. /// Extracted as a free function so tests can cover schema shape without a -/// live CommunicationManager harness. +/// live broadcast harness. fn build_fusion_audit_detail(src_ip: &str, attack_type: &str, fused: f32, per_source: &[SourceSample]) -> String { let per_source_json: Vec = per_source .iter() @@ -381,8 +379,7 @@ fn build_fusion_audit_detail(src_ip: &str, attack_type: &str, fused: f32, per_so #[cfg(test)] mod tests { - //! Orchestrator integration tests require an in-memory CommunicationManager - //! harness. Until then, the fusion math lives in `fusion_math::tests`, + //! Orchestrator unit tests. The fusion math lives in `fusion_math::tests`, //! canonical translation in `model::detection::attack_type::tests`, and //! the audit evidence schema is covered below. diff --git a/net-guardia/src/core/inference/traffic_logger.rs b/net-guardia/src/core/inference/traffic_logger.rs index c45965f..f66932b 100644 --- a/net-guardia/src/core/inference/traffic_logger.rs +++ b/net-guardia/src/core/inference/traffic_logger.rs @@ -6,7 +6,7 @@ //! //! On FIFO failure (permissions, I/O error) the writer shuts down //! cleanly, leaves inference untouched, logs through `MLLog`, and -//! (when a `CommunicationManager` is wired in) also publishes a WORM +//! (when an `audit_tx` sender is wired in) also publishes a WORM //! `AuditEvent` so the chain records an auditor-visible reason the //! recording stopped, not just a tracing line that may be lost. //! Callers see the channel disconnect and stop sending rows. @@ -21,12 +21,12 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use crossbeam::channel::{Receiver, Sender, TrySendError, bounded}; use macros::log; +use tokio::sync::broadcast; use crate::domain::common::config::constants::{FLOW_TRACE_FILE_EXT, FLOW_TRACE_FILE_MARKER}; use crate::domain::common::event::AuditEvent; use crate::domain::detection::error::MLError; use crate::domain::detection::log::MLLog; -use crate::infrastructure::communication_manager::CommunicationManager; pub const DEFAULT_MAX_FILE_BYTES: u64 = 500 * 1024 * 1024; pub const DEFAULT_MAX_FILE_AGE: Duration = Duration::from_secs(3600); @@ -69,7 +69,7 @@ pub struct TrafficLogger { impl TrafficLogger { /// Build a rotating writer rooted at `base_path`'s parent. Any /// existing `flow-trace-*.csv` in that directory participates in - /// the FIFO budget. `comm` is optional so tests (and paths where + /// the FIFO budget. `audit_tx` is optional so tests (and paths where /// the bus isn't wired yet) can exercise the rotation logic without /// the event-bus dependency; production code always passes `Some`. pub fn new( @@ -77,7 +77,7 @@ impl TrafficLogger { header: Vec, policy: RotationPolicy, channel_capacity: usize, - comm: Option>, + audit_tx: Option>, ) -> Result { let directory = base_path .parent() @@ -93,7 +93,7 @@ impl TrafficLogger { thread::Builder::new() .name("traffic-logger".to_string()) .spawn(move || { - writer_loop(receiver, writer_dir, writer_header, writer_policy, comm); + writer_loop(receiver, writer_dir, writer_header, writer_policy, audit_tx); })?; Ok(Self { @@ -188,14 +188,14 @@ fn writer_loop( directory: PathBuf, header: Vec, policy: RotationPolicy, - comm: Option>, + audit_tx: Option>, ) { let mut active = match open_new_file(&directory, &header) { Ok(a) => a, Err(e) => { let reason = e.to_string(); log!(MLLog::FlowTraceStopped(reason.clone())); - emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory); + emit_flow_trace_stop_audit(audit_tx.as_ref(), &reason, &directory); return; } }; @@ -213,7 +213,7 @@ fn writer_loop( // see the disconnect and stop trying. let reason = format!("FIFO sweep failed: {e}"); log!(MLLog::FlowTraceStopped(reason.clone())); - emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory); + emit_flow_trace_stop_audit(audit_tx.as_ref(), &reason, &directory); return; } active = match open_new_file(&directory, &header) { @@ -221,7 +221,7 @@ fn writer_loop( Err(e) => { let reason = format!("rotate failed: {e}"); log!(MLLog::FlowTraceStopped(reason.clone())); - emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory); + emit_flow_trace_stop_audit(audit_tx.as_ref(), &reason, &directory); return; } }; @@ -242,14 +242,14 @@ fn writer_loop( /// Publish a WORM `flow_trace_stopped` audit entry so an auditor can /// later see why Flow Trace recording went dormant without grepping -/// process logs. A `None` bus (tests, or pre-wiring paths) is a +/// process logs. A `None` sender (tests, or pre-wiring paths) is a /// deliberate no-op — the sibling `MLLog::FlowTraceStopped` tracing /// line still fires in both cases. /// /// Extracted as a free function so tests can cover the emit path /// without driving a full writer-loop + failing-disk fixture. -fn emit_flow_trace_stop_audit(comm: Option<&Arc>, reason: &str, directory: &Path) { - let Some(c) = comm else { +fn emit_flow_trace_stop_audit(audit_tx: Option<&broadcast::Sender>, reason: &str, directory: &Path) { + let Some(tx) = audit_tx else { return; }; let detail = serde_json::json!({ @@ -257,7 +257,7 @@ fn emit_flow_trace_stop_audit(comm: Option<&Arc>, reason: "directory": directory.display().to_string(), }) .to_string(); - let _ = c.publish_event_sync(AuditEvent { + let _ = tx.send(AuditEvent { actor: AUDIT_ACTOR_SYSTEM.to_string(), action: AUDIT_ACTION_FLOW_TRACE_STOPPED.to_string(), detail, @@ -401,13 +401,11 @@ mod tests { } #[tokio::test] - async fn stop_audit_reaches_subscriber_when_comm_provided() { - let comm = Arc::new(CommunicationManager::new(256)); - comm.register_event_type::(); - let mut rx = comm.subscribe_event::().unwrap(); + async fn stop_audit_reaches_subscriber_when_sender_provided() { + let (tx, mut rx) = broadcast::channel::(256); let dir = scratch_dir("audit-emit"); - emit_flow_trace_stop_audit(Some(&comm), "FIFO sweep failed: perm denied", &dir); + emit_flow_trace_stop_audit(Some(&tx), "FIFO sweep failed: perm denied", &dir); let event = rx.recv().await.expect("audit event must be delivered"); assert_eq!(event.actor, "system"); @@ -419,7 +417,7 @@ mod tests { } #[test] - fn stop_audit_noop_when_comm_absent() { + fn stop_audit_noop_when_sender_absent() { // Passing None is the explicit test-mode path — must not panic. emit_flow_trace_stop_audit(None, "any reason", Path::new("/tmp/anywhere")); } diff --git a/net-guardia/src/core/response/engine.rs b/net-guardia/src/core/response/engine.rs index 69e5e57..5a1a1b2 100644 --- a/net-guardia/src/core/response/engine.rs +++ b/net-guardia/src/core/response/engine.rs @@ -15,11 +15,9 @@ use crate::domain::common::error::Error; use crate::domain::common::event::ThreatDetectedEvent; use crate::domain::detection::attack_type::canonical_from_str; use crate::domain::response::condition::{ConditionType, PlaybookCondition}; -use crate::domain::response::error::SoarError; use crate::domain::response::log::SoarLog; use crate::domain::response::matcher::PlaybookMatcher; use crate::domain::response::playbook::{Playbook, PlaybookAction}; -use crate::infrastructure::communication_manager::CommunicationManager; use crate::interface::port::access_control::AccessControlPort; use crate::interface::port::app_repo::AppRepo; use crate::interface::port::geo_lookup::GeoLookup; @@ -202,19 +200,10 @@ impl SoarEngine { } /// Subscribe to ThreatDetectedEvent and start processing. - /// Returns an error if subscription fails — caller must handle this as a critical failure. - pub fn start(self: Arc, comm: Arc) -> Result<(), Error> { - let rx = comm.subscribe_event::().map_err(|e| { - log!(SoarLog::EventHandlingFailed(format!( - "CRITICAL: SOAR engine failed to subscribe — automated threat response is DISABLED: {}", - e - ))); - SoarError::ActionFailed("subscribe", e) - })?; + pub fn start(self: Arc, threat_rx: broadcast::Receiver) { tokio::spawn(async move { - Self::event_loop(self, rx).await; + Self::event_loop(self, threat_rx).await; }); - Ok(()) } async fn event_loop(self: Arc, mut rx: broadcast::Receiver) { diff --git a/net-guardia/src/domain/common/config/constants.rs b/net-guardia/src/domain/common/config/constants.rs index 20be277..12a88b2 100644 --- a/net-guardia/src/domain/common/config/constants.rs +++ b/net-guardia/src/domain/common/config/constants.rs @@ -30,3 +30,6 @@ pub const AUDIT_ACTOR_SECURITY_ADMIN_PREFIX: &str = "SecurityAdmin"; // ── Flow Trace ───────────────────────────────────────────────────── pub const FLOW_TRACE_FILE_MARKER: &str = "flow-trace-"; pub const FLOW_TRACE_FILE_EXT: &str = ".csv"; + +// ── Event Channels ──────────────────────────────────────────────── +pub const EVENT_CHANNEL_CAPACITY: usize = 256; diff --git a/net-guardia/src/domain/common/error/system.rs b/net-guardia/src/domain/common/error/system.rs index 33ad2e1..8571dd4 100644 --- a/net-guardia/src/domain/common/error/system.rs +++ b/net-guardia/src/domain/common/error/system.rs @@ -35,9 +35,6 @@ traceable! { #[error("Failed to set user groups")] SetUserGroupsFailed => tracing::Level::WARN, - #[error("Failed to bridge ML alert to SOAR")] - MlSoarBridgeFailed => tracing::Level::WARN, - #[error("Failed to store XDP mode")] XdpModeStoreFailed => tracing::Level::WARN, diff --git a/net-guardia/src/domain/common/event.rs b/net-guardia/src/domain/common/event.rs index 7cf48f4..cb9cb1c 100644 --- a/net-guardia/src/domain/common/event.rs +++ b/net-guardia/src/domain/common/event.rs @@ -52,7 +52,7 @@ impl FromStr for DetectionSource { // -- Detection Event (internal pipeline) -------------------------------------- /// Raw detection from any source. Sent via mpsc channel to DetectionOrchestrator. -/// Not published through CommunicationManager — this is a private internal pipeline. +/// Not published through broadcast — this is a private internal pipeline. #[derive(Debug, Clone)] pub struct DetectionEvent { pub source: DetectionSource, diff --git a/net-guardia/src/domain/common/log/audit.rs b/net-guardia/src/domain/common/log/audit.rs index 8a269a5..ef350f0 100644 --- a/net-guardia/src/domain/common/log/audit.rs +++ b/net-guardia/src/domain/common/log/audit.rs @@ -20,8 +20,5 @@ loggable! { #[error("AuditLogger: event channel closed")] AuditChannelClosed => tracing::Level::INFO, - - #[error("AuditLogger: failed to subscribe to events")] - AuditSubscribeFailed => tracing::Level::WARN, } } diff --git a/net-guardia/src/domain/common/log/http.rs b/net-guardia/src/domain/common/log/http.rs index 2809636..9f27e9f 100644 --- a/net-guardia/src/domain/common/log/http.rs +++ b/net-guardia/src/domain/common/log/http.rs @@ -11,8 +11,5 @@ 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/domain/detection/log.rs b/net-guardia/src/domain/detection/log.rs index bad2880..5f761de 100644 --- a/net-guardia/src/domain/detection/log.rs +++ b/net-guardia/src/domain/detection/log.rs @@ -127,12 +127,6 @@ loggable! { #[error("Fusion: window evicted under LRU pressure ({key_src} {key_type})")] FusionWindowEvicted { key_src: String, key_type: String } => tracing::Level::WARN, - #[error("Fusion: failed to publish ThreatDetectedEvent: {err}")] - FusionPublishFailed { err: String } => tracing::Level::ERROR, - - #[error("Fusion: failed to publish AuditEvent: {err}")] - FusionAuditPublishFailed { err: String } => tracing::Level::ERROR, - #[error("Detection event dropped (channel full): {detector} {attack_type} from {source_ip}")] DetectionChannelDrop { detector: String, attack_type: String, source_ip: String } => tracing::Level::WARN, } diff --git a/net-guardia/src/infrastructure/app_services.rs b/net-guardia/src/infrastructure/app_services.rs index d97daf5..9604f46 100644 --- a/net-guardia/src/infrastructure/app_services.rs +++ b/net-guardia/src/infrastructure/app_services.rs @@ -5,6 +5,7 @@ use std::time::{Duration, SystemTime}; use arc_swap::ArcSwap; use crossbeam::queue::SegQueue; use macros::log; +use tokio::sync::broadcast; use tokio::sync::oneshot; use crate::core::inference::alert::MLAlert; @@ -18,6 +19,7 @@ use crate::domain::common::config::constants::{MANIFEST_FILENAME, MODELS_DIR}; use crate::domain::common::error::Error; use crate::domain::common::error::misc::MiscError; use crate::domain::common::error::system::SystemError; +use crate::domain::common::event::AuditEvent; use crate::domain::common::log::system::SystemLog; use crate::domain::common::system::health::EbpfHealth; use crate::domain::detection::flow_features::FlowFeatures; @@ -29,7 +31,6 @@ use crate::domain::detection::ml_detection::EngineConfig; use crate::domain::detection::ml_inference_config::MLInferenceConfig; use crate::domain::detection::model_adapter::ModelSourceState; use crate::domain::detection::model_source::ModelInfo; -use crate::infrastructure::communication_manager::CommunicationManager; use crate::infrastructure::health::SystemHealth; use crate::infrastructure::statistics::FlowStatistics; @@ -53,7 +54,7 @@ impl AppServices { ml_manifest: Option, drift_detector: DriftDetectorHandle, ebpf_health: Arc>, - comm: Arc, + audit_tx: broadcast::Sender, ) -> Result { let health = SystemHealth::new(app_config.clone(), ebpf_health)?; @@ -126,7 +127,7 @@ impl AppServices { header, policy, config.ml.traffic_logger_channel_capacity, - Some(comm.clone()), + Some(audit_tx.clone()), ) .map_err(|e| MiscError::TrafficLogCreateError(csv_path.clone(), e.to_string()))?; log!(SystemLog::TrafficLoggingEnabled(csv_path)); diff --git a/net-guardia/src/infrastructure/audit_logger.rs b/net-guardia/src/infrastructure/audit_logger.rs index da81ed0..8e14f37 100644 --- a/net-guardia/src/infrastructure/audit_logger.rs +++ b/net-guardia/src/infrastructure/audit_logger.rs @@ -1,11 +1,11 @@ use std::sync::Arc; use macros::log; +use tokio::sync::broadcast; use tokio::sync::broadcast::error::RecvError; use crate::domain::common::event::{AuditEvent, DriftDetectedEvent}; use crate::domain::common::log::audit::AuditLog; -use crate::infrastructure::communication_manager::CommunicationManager; use crate::interface::port::audit::AuditRepo; /// Subscribes to `AuditEvent` and persists each entry to the `audit_log` table. @@ -19,12 +19,16 @@ impl AuditLogger { Self { db } } - /// Subscribe to AuditEvent and DriftDetectedEvent on the communication manager - /// and start background tasks that persist events to DB + structured logs. - pub fn start(self: Arc, comm: &CommunicationManager) { - // Subscribe to AuditEvent - if let Ok(mut rx) = comm.subscribe_event::() { + /// Start background tasks that persist audit and drift events to DB + structured logs. + pub fn start( + self: Arc, + audit_rx: broadcast::Receiver, + drift_rx: broadcast::Receiver, + ) { + // Drain AuditEvent receiver + { let this = self.clone(); + let mut rx = audit_rx; tokio::spawn(async move { loop { match rx.recv().await { @@ -41,13 +45,12 @@ impl AuditLogger { } } }); - } else { - log!(AuditLog::AuditSubscribeFailed); } - // Subscribe to DriftDetectedEvent — log as audit trail entry - if let Ok(mut rx) = comm.subscribe_event::() { + // Drain DriftDetectedEvent receiver — log as audit trail entry + { let this = self; + let mut rx = drift_rx; tokio::spawn(async move { loop { match rx.recv().await { @@ -64,8 +67,6 @@ impl AuditLogger { } } }); - } else { - log!(AuditLog::AuditSubscribeFailed); } } diff --git a/net-guardia/src/infrastructure/communication_manager.rs b/net-guardia/src/infrastructure/communication_manager.rs deleted file mode 100644 index 186133b..0000000 --- a/net-guardia/src/infrastructure/communication_manager.rs +++ /dev/null @@ -1,373 +0,0 @@ -//! Cross-BC in-process event bus (technical service, not a BC). -//! -//! Per `docs/strategy/DOMAIN_MAP.md` §2, Communication Bus is a Technical -//! Service — it has no ubiquitous language, no domain expert, no aggregate. -//! It stays in `infrastructure/` and never takes a BC folder name. The trait -//! surface (`Event`, `Command`, `Query`, `EventBroadcaster`, `CommandHandler`) -//! lives at `interface/communication/` and remains untouched. - -use std::any::{Any, TypeId}; -use std::sync::Arc; - -use dashmap::DashMap; -use tokio::sync::broadcast; - -use crate::domain::common::error::Error; -use crate::domain::common::error::misc::MiscError; -use crate::interface::communication::command::*; -use crate::interface::communication::event::Event; -use crate::interface::communication::event::EventBroadcaster; -use crate::interface::communication::query::*; - -/// Inline TypedEventBroadcaster (adapted from MirrorSphere's model). -pub struct TypedEventBroadcaster { - pub sender: broadcast::Sender, -} - -impl EventBroadcaster for TypedEventBroadcaster { - fn subscribe_typed(&self) -> Box { - Box::new(self.sender.subscribe()) - } - - fn broadcast_event(&self, event: Box) -> Result<(), Error> { - let typed_event = *event.downcast::().map_err(|_| MiscError::TypeMismatch)?; - let _ = self.sender.send(typed_event); - Ok(()) - } -} - -/// Central communication hub using the command/query/event pattern. -/// Adapted from MirrorSphere's CommunicationManager for NetGuardia. -pub struct CommunicationManager { - command_handlers: DashMap, - query_handlers: DashMap, - event_broadcasters: DashMap>, - channel_capacity: usize, -} - -impl CommunicationManager { - pub fn new(channel_capacity: usize) -> Self { - Self { - command_handlers: DashMap::new(), - query_handlers: DashMap::new(), - event_broadcasters: DashMap::new(), - channel_capacity: channel_capacity.max(1), - } - } - - pub fn with_service(self: Arc, service: Arc) -> ServiceRegistrar { - ServiceRegistrar::new(service, self) - } - - pub fn register_command_handler(&self, handler: Arc + Send + Sync>) { - let type_id = TypeId::of::(); - let boxed_handler: CommandHandlerFn = Box::new(move |command: Box| { - let handler = handler.clone(); - Box::pin(async move { - let command = *command.downcast::().map_err(|_| MiscError::TypeMismatch)?; - handler.handle_command(command).await - }) as CommandFuture - }); - - self.command_handlers.insert(type_id, boxed_handler); - } - - pub async fn send_command(&self, command: C) -> Result<(), Error> { - let type_id = TypeId::of::(); - if let Some(handler) = self.command_handlers.get(&type_id) { - handler(Box::new(command)).await - } else { - Err(MiscError::HandlerNotFound)? - } - } - - pub fn register_query_handler(&self, handler: Arc + Send + Sync>) { - let type_id = TypeId::of::(); - let boxed_handler: QueryHandlerFn = Box::new(move |query: Box| { - let handler = handler.clone(); - Box::pin(async move { - let query = *query.downcast::().map_err(|_| MiscError::TypeMismatch)?; - let response = handler.handle_query(query).await?; - Ok(Box::new(response) as Box) - }) as QueryFuture - }); - - self.query_handlers.insert(type_id, boxed_handler); - } - - pub async fn send_query(&self, query: Q) -> Result { - let type_id = TypeId::of::(); - if let Some(handler) = self.query_handlers.get(&type_id) { - let response = handler(Box::new(query)).await?; - Ok(*response - .downcast::() - .map_err(|_| MiscError::TypeMismatch)?) - } else { - Err(MiscError::HandlerNotFound)? - } - } - - pub fn register_event_type(&self) { - let type_id = TypeId::of::(); - let (tx, _) = broadcast::channel::(self.channel_capacity); - let broadcaster = TypedEventBroadcaster { sender: tx }; - self.event_broadcasters.insert(type_id, Box::new(broadcaster)); - } - - pub fn subscribe_event(&self) -> Result, Error> { - let type_id = TypeId::of::(); - let broadcaster = self - .event_broadcasters - .get(&type_id) - .ok_or(MiscError::TypeNotRegistered)?; - let receiver_box = broadcaster.subscribe_typed(); - let receiver = *receiver_box - .downcast::>() - .map_err(|_| MiscError::TypeMismatch)?; - Ok(receiver) - } - - pub async fn publish_event(&self, event: E) -> Result<(), Error> { - let type_id = TypeId::of::(); - let broadcaster = self - .event_broadcasters - .get(&type_id) - .ok_or(MiscError::TypeNotRegistered)?; - broadcaster.broadcast_event(Box::new(event)) - } - - /// Synchronous counterpart for callers that live outside the tokio - /// runtime — in particular, the Flow Trace writer thread, which - /// runs on a dedicated `std::thread` and can't `.await`. The - /// internal broadcast channel is already non-blocking, so the - /// `async fn` sibling never actually yields; this variant exposes - /// the same work without the ceremony. - pub fn publish_event_sync(&self, event: E) -> Result<(), Error> { - let type_id = TypeId::of::(); - let broadcaster = self - .event_broadcasters - .get(&type_id) - .ok_or(MiscError::TypeNotRegistered)?; - broadcaster.broadcast_event(Box::new(event)) - } -} - -/// Fluent builder for registering a service's command/query/event handlers. -pub struct ServiceRegistrar { - service: Arc, - comm: Arc, -} - -impl ServiceRegistrar { - fn new(service: Arc, comm: Arc) -> Self { - Self { service, comm } - } - - pub fn command(self) -> Self - where - S: CommandHandler, - { - let handler: Arc + Send + Sync> = self.service.clone(); - self.comm.register_command_handler::(handler); - self - } - - pub fn query(self) -> Self - where - S: QueryHandler, - { - let handler: Arc + Send + Sync> = self.service.clone(); - self.comm.register_query_handler::(handler); - self - } - - pub fn build(self) -> Arc { - self.comm - } -} - -#[cfg(test)] -mod tests { - use std::sync::Mutex; - - use async_trait::async_trait; - - use super::*; - use crate::interface::communication::command::Command; - use crate::interface::communication::event::Event; - use crate::interface::communication::message::Message; - use crate::interface::communication::query::Query; - - // ── Test Command ───────────────────────────────────────────────── - - struct TestCommand { - value: String, - } - - impl Message for TestCommand { - type Response = (); - } - impl Command for TestCommand {} - - struct TestCommandHandler { - received: Arc>>, - } - - #[async_trait] - impl CommandHandler for TestCommandHandler { - async fn handle_command(&self, command: TestCommand) -> Result<(), Error> { - self.received.lock().unwrap().push(command.value); - Ok(()) - } - } - - // ── Test Query ─────────────────────────────────────────────────── - - struct TestQuery { - input: i32, - } - - impl Message for TestQuery { - type Response = i32; - } - impl Query for TestQuery {} - - struct TestQueryHandler; - - #[async_trait] - impl QueryHandler for TestQueryHandler { - async fn handle_query(&self, query: TestQuery) -> Result { - Ok(query.input * 2) - } - } - - // ── Test Event ─────────────────────────────────────────────────── - - #[derive(Debug, Clone)] - struct TestEvent { - message: String, - } - impl Event for TestEvent {} - - // ── Tests ──────────────────────────────────────────────────────── - - #[tokio::test] - async fn test_command_dispatch() { - let received = Arc::new(Mutex::new(Vec::new())); - let handler = Arc::new(TestCommandHandler { - received: received.clone(), - }); - - let comm = Arc::new(CommunicationManager::new(256)); - comm.register_command_handler::(handler); - - comm.send_command(TestCommand { value: "hello".into() }).await.unwrap(); - - let msgs = received.lock().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0], "hello"); - } - - #[tokio::test] - async fn test_command_not_found() { - let comm = CommunicationManager::new(256); - let result = comm.send_command(TestCommand { value: "nope".into() }).await; - assert!(result.is_err()); - } - - #[tokio::test] - async fn test_query_dispatch() { - let handler = Arc::new(TestQueryHandler); - let comm = Arc::new(CommunicationManager::new(256)); - comm.register_query_handler::(handler); - - let result = comm.send_query(TestQuery { input: 21 }).await.unwrap(); - assert_eq!(result, 42); - } - - #[tokio::test] - async fn test_query_not_found() { - let comm = CommunicationManager::new(256); - let result = comm.send_query(TestQuery { input: 1 }).await; - assert!(result.is_err()); - } - - #[tokio::test] - async fn test_event_pub_sub() { - let comm = CommunicationManager::new(256); - comm.register_event_type::(); - - let mut receiver = comm.subscribe_event::().unwrap(); - - comm.publish_event(TestEvent { message: "ping".into() }).await.unwrap(); - - let event = receiver.recv().await.unwrap(); - assert_eq!(event.message, "ping"); - } - - #[tokio::test] - async fn test_event_not_registered() { - let comm = CommunicationManager::new(256); - let result = comm.subscribe_event::(); - assert!(result.is_err()); - } - - #[tokio::test] - async fn publish_event_sync_delivers_to_subscriber() { - let comm = CommunicationManager::new(256); - comm.register_event_type::(); - let mut rx = comm.subscribe_event::().unwrap(); - - comm.publish_event_sync(TestEvent { message: "sync".into() }).unwrap(); - - let event = rx.recv().await.unwrap(); - assert_eq!(event.message, "sync"); - } - - #[test] - fn publish_event_sync_errors_when_type_unregistered() { - let comm = CommunicationManager::new(256); - let result = comm.publish_event_sync(TestEvent { - message: "dropped".into(), - }); - assert!(result.is_err()); - } - - #[tokio::test] - async fn test_event_multiple_subscribers() { - let comm = CommunicationManager::new(256); - comm.register_event_type::(); - - let mut rx1 = comm.subscribe_event::().unwrap(); - let mut rx2 = comm.subscribe_event::().unwrap(); - - comm.publish_event(TestEvent { - message: "broadcast".into(), - }) - .await - .unwrap(); - - assert_eq!(rx1.recv().await.unwrap().message, "broadcast"); - assert_eq!(rx2.recv().await.unwrap().message, "broadcast"); - } - - #[tokio::test] - async fn test_service_registrar() { - let received = Arc::new(Mutex::new(Vec::new())); - let handler = Arc::new(TestCommandHandler { - received: received.clone(), - }); - - let comm = Arc::new(CommunicationManager::new(256)); - let _comm = comm.clone().with_service(handler).command::().build(); - - comm.send_command(TestCommand { - value: "via_registrar".into(), - }) - .await - .unwrap(); - - let msgs = received.lock().unwrap(); - assert_eq!(msgs[0], "via_registrar"); - } -} diff --git a/net-guardia/src/infrastructure/enforce_mode_handler.rs b/net-guardia/src/infrastructure/enforce_mode_handler.rs index 2f937bf..c24975e 100644 --- a/net-guardia/src/infrastructure/enforce_mode_handler.rs +++ b/net-guardia/src/infrastructure/enforce_mode_handler.rs @@ -1,17 +1,12 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU8, Ordering}; -use async_trait::async_trait; use macros::log; +use tokio::sync::broadcast; use crate::domain::common::error::Error; use crate::domain::common::event::AuditEvent; use crate::domain::common::log::system::SystemLog; -use crate::infrastructure::communication_manager::CommunicationManager; -use crate::interface::communication::command::CommandHandler; -use crate::interface::communication::command_types::ChangeEnforceModeCommand; -use crate::interface::communication::query::QueryHandler; -use crate::interface::communication::query_types::GetEnforceModeQuery; use crate::interface::port::app_repo::AppRepo; /// Map enforce-mode string to u8: monitor=0, ml_only=1, enforce=2. @@ -26,46 +21,36 @@ pub fn enforce_mode_to_u8(mode: &str) -> u8 { /// Handles enforce-mode commands and queries by delegating to the repository. pub struct EnforceModeHandler { db: Arc, - comm: Arc, + audit_tx: broadcast::Sender, /// Shared AtomicU8 cache: Monitor=0, MlOnly=1, Enforce=2. enforce_cache: Arc, } impl EnforceModeHandler { - pub fn new(db: Arc, comm: Arc, enforce_cache: Arc) -> Self { + pub fn new(db: Arc, audit_tx: broadcast::Sender, enforce_cache: Arc) -> Self { Self { db, - comm, + audit_tx, enforce_cache, } } -} -#[async_trait] -impl CommandHandler for EnforceModeHandler { - async fn handle_command(&self, command: ChangeEnforceModeCommand) -> Result<(), Error> { - self.db.set_setting("enforce_mode", &command.mode)?; - self.enforce_cache - .store(enforce_mode_to_u8(&command.mode), Ordering::SeqCst); - log!(SystemLog::EnforceModeChanged(command.mode.clone())); + pub fn change_mode(&self, mode: String) -> Result<(), Error> { + self.db.set_setting("enforce_mode", &mode)?; + self.enforce_cache.store(enforce_mode_to_u8(&mode), Ordering::SeqCst); + log!(SystemLog::EnforceModeChanged(mode.clone())); // Publish audit event for the mode change - let _ = self - .comm - .publish_event(AuditEvent { - actor: "admin".to_string(), - action: "enforce_mode_changed".to_string(), - detail: serde_json::json!({ "new_mode": command.mode }).to_string(), - }) - .await; + let _ = self.audit_tx.send(AuditEvent { + actor: "admin".to_string(), + action: "enforce_mode_changed".to_string(), + detail: serde_json::json!({ "new_mode": mode }).to_string(), + }); Ok(()) } -} -#[async_trait] -impl QueryHandler for EnforceModeHandler { - async fn handle_query(&self, _query: GetEnforceModeQuery) -> Result { + pub fn get_mode(&self) -> Result { match self.db.get_setting("enforce_mode")? { Some(mode) => Ok(mode), None => Ok("monitor".to_string()), @@ -77,73 +62,53 @@ impl QueryHandler for EnforceModeHandler { mod tests { use super::*; use crate::adapter::persistence::Database; - use crate::infrastructure::communication_manager::CommunicationManager; - use crate::interface::communication::command_types::ChangeEnforceModeCommand; - use crate::interface::communication::query_types::GetEnforceModeQuery; + use crate::domain::common::config::constants::EVENT_CHANNEL_CAPACITY; - fn test_handler() -> (Arc, Arc) { + fn test_handler() -> EnforceModeHandler { let db = Arc::new(Database::new(":memory:").unwrap()) as Arc; let cache = Arc::new(AtomicU8::new(0)); - let comm = Arc::new(CommunicationManager::new(256)); - comm.register_event_type::(); - let handler = Arc::new(EnforceModeHandler::new(db, comm.clone(), cache)); - let _ = comm - .clone() - .with_service(handler.clone()) - .command::() - .query::() - .build(); - (handler, comm) + let (audit_tx, _rx) = broadcast::channel::(EVENT_CHANNEL_CAPACITY); + EnforceModeHandler::new(db, audit_tx, cache) } - #[tokio::test] - async fn test_default_mode_is_monitor() { - let (_, comm) = test_handler(); - let mode = comm.send_query(GetEnforceModeQuery).await.unwrap(); + #[test] + fn test_default_mode_is_monitor() { + let handler = test_handler(); + let mode = handler.get_mode().unwrap(); assert_eq!(mode, "monitor"); } - #[tokio::test] - async fn test_change_to_enforce() { - let (_, comm) = test_handler(); - comm.send_command(ChangeEnforceModeCommand { mode: "enforce".into() }) - .await - .unwrap(); - let mode = comm.send_query(GetEnforceModeQuery).await.unwrap(); + #[test] + fn test_change_to_enforce() { + let handler = test_handler(); + handler.change_mode("enforce".into()).unwrap(); + let mode = handler.get_mode().unwrap(); assert_eq!(mode, "enforce"); } - #[tokio::test] - async fn test_change_back_to_monitor() { - let (_, comm) = test_handler(); - comm.send_command(ChangeEnforceModeCommand { mode: "enforce".into() }) - .await - .unwrap(); - comm.send_command(ChangeEnforceModeCommand { mode: "monitor".into() }) - .await - .unwrap(); - let mode = comm.send_query(GetEnforceModeQuery).await.unwrap(); + #[test] + fn test_change_back_to_monitor() { + let handler = test_handler(); + handler.change_mode("enforce".into()).unwrap(); + handler.change_mode("monitor".into()).unwrap(); + let mode = handler.get_mode().unwrap(); assert_eq!(mode, "monitor"); } - #[tokio::test] - async fn test_change_to_ml_only() { - let (_, comm) = test_handler(); - comm.send_command(ChangeEnforceModeCommand { mode: "ml_only".into() }) - .await - .unwrap(); - let mode = comm.send_query(GetEnforceModeQuery).await.unwrap(); + #[test] + fn test_change_to_ml_only() { + let handler = test_handler(); + handler.change_mode("ml_only".into()).unwrap(); + let mode = handler.get_mode().unwrap(); assert_eq!(mode, "ml_only"); } - #[tokio::test] - async fn test_cycle_all_modes() { - let (_, comm) = test_handler(); + #[test] + fn test_cycle_all_modes() { + let handler = test_handler(); for mode_str in ["enforce", "ml_only", "monitor"] { - comm.send_command(ChangeEnforceModeCommand { mode: mode_str.into() }) - .await - .unwrap(); - let mode = comm.send_query(GetEnforceModeQuery).await.unwrap(); + handler.change_mode(mode_str.into()).unwrap(); + let mode = handler.get_mode().unwrap(); assert_eq!(mode, mode_str); } } diff --git a/net-guardia/src/infrastructure/http_server.rs b/net-guardia/src/infrastructure/http_server.rs index 771fc32..6a94ed1 100644 --- a/net-guardia/src/infrastructure/http_server.rs +++ b/net-guardia/src/infrastructure/http_server.rs @@ -8,6 +8,7 @@ use actix_web::web::route; use actix_web::{App, HttpResponse, HttpServer, web}; use arc_swap::ArcSwap; use macros::log; +use tokio::sync::broadcast; use crate::adapter::ebpf::EbpfServices; use crate::adapter::http::model_upload::PromoteGate; @@ -35,11 +36,12 @@ use crate::domain::common::config::AppConfig; use crate::domain::common::config::constants::HTTP_FALLBACK_PORT; use crate::domain::common::error::Error; use crate::domain::common::error::http::HttpError; +use crate::domain::common::event::{AuditEvent, ThreatDetectedEvent}; use crate::domain::common::log::http::HttpLog; use crate::domain::common::system::readiness::ReadinessState; use crate::domain::detection::ml_inference_config::MLInferenceConfig; use crate::infrastructure::app_services::AppServices; -use crate::infrastructure::communication_manager::CommunicationManager; +use crate::infrastructure::enforce_mode_handler::EnforceModeHandler; use crate::infrastructure::logger::Logger; use crate::infrastructure::secret_store::SecretStore; use crate::infrastructure::suricata_manager::SuricataManager; @@ -60,7 +62,9 @@ pub struct HttpServerParams { pub db: Arc, pub secret_store: Arc, pub jwt_service: Arc, - pub comm: Arc, + pub threat_tx: broadcast::Sender, + pub audit_tx: broadcast::Sender, + pub enforce_handler: Arc, pub setup_complete: SetupCompleteFlag, pub ready: ReadyFlag, pub readiness_state: Arc, @@ -228,7 +232,9 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> { let db = params.db; let secret_store = params.secret_store; let jwt_service = params.jwt_service; - let comm = params.comm; + let threat_tx = params.threat_tx; + let audit_tx = params.audit_tx; + let enforce_handler = params.enforce_handler; let setup_complete = params.setup_complete; let ready = params.ready; let readiness_state = params.readiness_state; @@ -278,7 +284,9 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> { .app_data(web::Data::from(db.clone())) .app_data(web::Data::from(secret_store.clone())) .app_data(web::Data::from(jwt_service.clone())) - .app_data(web::Data::from(comm.clone())) + .app_data(web::Data::new(threat_tx.clone())) + .app_data(web::Data::new(audit_tx.clone())) + .app_data(web::Data::from(enforce_handler.clone())) .app_data(web::Data::new(setup_complete.clone())) .app_data(web::Data::new(ready.clone())) .app_data(web::Data::from(readiness_state.clone())) diff --git a/net-guardia/src/infrastructure/logging.rs b/net-guardia/src/infrastructure/logging.rs new file mode 100644 index 0000000..4eb4e4a --- /dev/null +++ b/net-guardia/src/infrastructure/logging.rs @@ -0,0 +1,200 @@ +use std::fs; +use std::sync::OnceLock; +use std::{env, io}; + +use tracing::Level; +use tracing::level_filters::LevelFilter; +use tracing_appender::rolling::{RollingFileAppender, Rotation}; +use tracing_subscriber::filter::Directive; +use tracing_subscriber::filter::EnvFilter; +use tracing_subscriber::fmt::layer as fmt_layer; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::util::SubscriberInitExt; +use tracing_subscriber::{Layer, filter, reload}; + +use crate::core::common::observability::log_buffer::LogBufferLayer; +use crate::domain::common::config::observability::ObservabilityConfig; +use crate::domain::common::error::Error; +use crate::domain::common::error::io::IOError; +use crate::interface::utils::logging::FilterControl; + +/// Type-erased reload handle stored as a trait object. +/// We erase the complex layered type by boxing the modify closure. +static FILTER_HANDLE: OnceLock> = OnceLock::new(); + +/// Snapshot of the per-target directives (e.g. `maxminddb=warn`) in effect +/// at `initialize()` time. `set_level` rebuilds the filter from scratch +/// around a new root level; reapplying these keeps any RUST_LOG overrides +/// the operator configured for specific crates from being silently lost. +static PRESERVED_DIRECTIVES: OnceLock> = OnceLock::new(); + +pub struct Logging; + +impl Logging { + pub fn initialize(config: &ObservabilityConfig) -> Result<(), Error> { + let log_directory = "logs"; + fs::create_dir_all(log_directory).map_err(|err| IOError::CreateDirectoryFailed(log_directory, err))?; + + let file_appender = RollingFileAppender::new(Rotation::DAILY, log_directory, "NetGuardia"); + + let stdout_layer = fmt_layer() + .with_file(true) + .with_line_number(true) + .with_thread_ids(true) + .with_target(false) + .with_ansi(true); + + let file_layer = fmt_layer() + .with_file(false) + .with_line_number(false) + .with_thread_ids(false) + .with_target(true) + .with_ansi(false) + .with_writer(file_appender); + + // RUST_LOG overrides the config value when present + let level: Level = env::var("RUST_LOG") + .ok() + .and_then(|s| s.parse().ok()) + .or_else(|| config.log_level.parse().ok()) + .unwrap_or(Level::INFO); + + let mut preserved: Vec = env::var("RUST_LOG") + .ok() + .map(|raw| { + raw.split(',') + .map(|s| s.trim().to_string()) + .filter(|d| !d.is_empty() && d.contains('=')) + .collect() + }) + .unwrap_or_default(); + if !preserved.iter().any(|d| d == "maxminddb=warn") { + preserved.push("maxminddb=warn".to_string()); + } + let _ = PRESERVED_DIRECTIVES.set(preserved); + + let mut filter = EnvFilter::new(level.to_string()); + if let Some(directives) = PRESERVED_DIRECTIVES.get() { + for d in directives { + if let Ok(parsed) = d.parse::() { + filter = filter.add_directive(parsed); + } + } + } + + let (filter_layer, reload_handle) = reload::Layer::new(filter); + + tracing_subscriber::registry() + .with(filter_layer) + .with(stdout_layer) + .with(file_layer) + .with(LogBufferLayer::new(config.log_buffer_capacity, config.log_buffer_max_message_bytes)) + .init(); + + let _ = FILTER_HANDLE.set(Box::new(reload_handle)); + + Ok(()) + } + + pub fn initialize_cli() -> Result<(), Error> { + let stdout_layer = fmt_layer() + .without_time() + .with_level(false) + .with_target(false) + .with_file(false) + .with_line_number(false) + .with_thread_ids(false) + .with_ansi(false) + .with_writer(io::stdout) + .with_filter(LevelFilter::INFO); + + let stderr_layer = fmt_layer() + .without_time() + .with_level(false) + .with_target(false) + .with_writer(io::stderr) + .with_filter(filter::filter_fn(|m| m.level() <= &Level::WARN)); + + tracing_subscriber::registry() + .with(stdout_layer) + .with(stderr_layer) + .init(); + Ok(()) + } + + pub fn set_level(level: &str) -> Result { + let handle = FILTER_HANDLE.get().ok_or("Logging not initialized")?; + + let parsed_level: Level = level.parse().map_err(|_| { + format!( + "Invalid log level '{}'. Valid levels: trace, debug, info, warn, error", + level + ) + })?; + + let mut new_filter = EnvFilter::new(parsed_level.to_string()); + if let Some(directives) = PRESERVED_DIRECTIVES.get() { + for d in directives { + if let Ok(parsed) = d.parse::() { + new_filter = new_filter.add_directive(parsed); + } + } + } + + handle.reload_filter(new_filter)?; + + Ok(parsed_level.to_string().to_lowercase()) + } + + /// 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 `