diff --git a/net-guardia/src/core/ml/traffic_logger.rs b/net-guardia/src/core/ml/traffic_logger.rs index 481da16..c1bc21a 100644 --- a/net-guardia/src/core/ml/traffic_logger.rs +++ b/net-guardia/src/core/ml/traffic_logger.rs @@ -5,7 +5,10 @@ //! recording session can't eat the disk. //! //! On FIFO failure (permissions, I/O error) the writer shuts down -//! cleanly, leaves inference untouched, and logs through `MLLog`. +//! cleanly, leaves inference untouched, logs through `MLLog`, and +//! (when a `CommunicationManager` 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. use std::fs::{File, OpenOptions}; @@ -18,7 +21,9 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use crossbeam::channel::{Receiver, Sender, TrySendError, bounded}; use macros::log; +use crate::infrastructure::communication_manager::CommunicationManager; use crate::model::error::ml::MLError; +use crate::model::event::AuditEvent; use crate::model::log::ml::MLLog; /// Default per-file size cap. A single CSV file won't grow past this @@ -72,8 +77,15 @@ 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. - pub fn new(base_path: &Path, header: Vec, policy: RotationPolicy) -> Result { + /// the FIFO budget. `comm` 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( + base_path: &Path, + header: Vec, + policy: RotationPolicy, + comm: Option>, + ) -> Result { let directory = base_path .parent() .map(Path::to_path_buf) @@ -88,7 +100,7 @@ impl TrafficLogger { thread::Builder::new() .name("traffic-logger".to_string()) .spawn(move || { - writer_loop(receiver, writer_dir, writer_header, writer_policy); + writer_loop(receiver, writer_dir, writer_header, writer_policy, comm); })?; Ok(Self { @@ -178,11 +190,19 @@ fn parse_timestamp_suffix(name: &str) -> Option { without_ext.parse::().ok() } -fn writer_loop(receiver: Receiver>, directory: PathBuf, header: Vec, policy: RotationPolicy) { +fn writer_loop( + receiver: Receiver>, + directory: PathBuf, + header: Vec, + policy: RotationPolicy, + comm: Option>, +) { let mut active = match open_new_file(&directory, &header) { Ok(a) => a, Err(e) => { - log!(MLLog::FlowTraceStopped(e.to_string())); + let reason = e.to_string(); + log!(MLLog::FlowTraceStopped(reason.clone())); + emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory); return; } }; @@ -198,13 +218,17 @@ fn writer_loop(receiver: Receiver>, directory: PathBuf, header: Vec< // FIFO failure is the documented "stop Flow Trace, keep // inference running" path. Drop the channel so callers // see the disconnect and stop trying. - log!(MLLog::FlowTraceStopped(format!("FIFO sweep failed: {e}"))); + let reason = format!("FIFO sweep failed: {e}"); + log!(MLLog::FlowTraceStopped(reason.clone())); + emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory); return; } active = match open_new_file(&directory, &header) { Ok(a) => a, Err(e) => { - log!(MLLog::FlowTraceStopped(format!("rotate failed: {e}"))); + let reason = format!("rotate failed: {e}"); + log!(MLLog::FlowTraceStopped(reason.clone())); + emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory); return; } }; @@ -223,6 +247,30 @@ fn writer_loop(receiver: Receiver>, directory: PathBuf, header: Vec< } } +/// 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 +/// 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 { + return; + }; + let detail = serde_json::json!({ + "reason": reason, + "directory": directory.display().to_string(), + }) + .to_string(); + let _ = c.publish_event_sync(AuditEvent { + actor: "system".to_string(), + action: "flow_trace_stopped".to_string(), + detail, + }); +} + struct ActiveFile { writer: BufWriter, opened_at: Instant, @@ -357,4 +405,28 @@ mod tests { let missing = PathBuf::from("/nonexistent/flow-trace/dir"); assert!(list_flow_trace_files(&missing).unwrap().is_empty()); } + + #[tokio::test] + async fn stop_audit_reaches_subscriber_when_comm_provided() { + let comm = Arc::new(CommunicationManager::new()); + comm.register_event_type::(); + let mut rx = comm.subscribe_event::().unwrap(); + + let dir = scratch_dir("audit-emit"); + emit_flow_trace_stop_audit(Some(&comm), "FIFO sweep failed: perm denied", &dir); + + let event = rx.recv().await.expect("audit event must be delivered"); + assert_eq!(event.actor, "system"); + assert_eq!(event.action, "flow_trace_stopped"); + let parsed: serde_json::Value = serde_json::from_str(&event.detail).unwrap(); + assert_eq!(parsed["reason"], "FIFO sweep failed: perm denied"); + assert_eq!(parsed["directory"], dir.display().to_string()); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn stop_audit_noop_when_comm_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/infrastructure/app_services.rs b/net-guardia/src/infrastructure/app_services.rs index ab02316..9d49fd7 100644 --- a/net-guardia/src/infrastructure/app_services.rs +++ b/net-guardia/src/infrastructure/app_services.rs @@ -16,6 +16,7 @@ use crate::core::ml::manifest::ModelManifest; use crate::core::ml::model_loader::build_adapter; use crate::core::ml::traffic_logger::{RotationPolicy, TrafficLogger}; use crate::infrastructure::app_config::AppConfig; +use crate::infrastructure::communication_manager::CommunicationManager; use crate::infrastructure::health::SystemHealth; use crate::infrastructure::statistics::FlowStatistics; use crate::model::config::constants::{MANIFEST_FILENAME, MODELS_DIR}; @@ -50,6 +51,7 @@ impl AppServices { ml_manifest: Option, drift_detector: Arc>, ebpf_health: Arc>, + comm: Arc, ) -> Result { let health = SystemHealth::new(app_config.clone(), ebpf_health)?; @@ -100,7 +102,7 @@ impl AppServices { let mut header = FlowFeatures::all_feature_names_owned(); header.push("Label".to_string()); let base_path = PathBuf::from(&csv_path); - let logger = TrafficLogger::new(&base_path, header, RotationPolicy::default()) + let logger = TrafficLogger::new(&base_path, header, RotationPolicy::default(), Some(comm.clone())) .map_err(|e| MiscError::TrafficLogCreateError(csv_path.clone(), e.to_string()))?; log!(SystemLog::TrafficLoggingEnabled(csv_path)); Some(Arc::new(logger)) diff --git a/net-guardia/src/infrastructure/communication_manager.rs b/net-guardia/src/infrastructure/communication_manager.rs index cb9ae05..526433e 100644 --- a/net-guardia/src/infrastructure/communication_manager.rs +++ b/net-guardia/src/infrastructure/communication_manager.rs @@ -134,6 +134,21 @@ impl CommunicationManager { .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. @@ -295,6 +310,27 @@ mod tests { assert!(result.is_err()); } + #[tokio::test] + async fn publish_event_sync_delivers_to_subscriber() { + let comm = CommunicationManager::new(); + 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(); + 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(); diff --git a/net-guardia/src/infrastructure/service_factory.rs b/net-guardia/src/infrastructure/service_factory.rs index 19922cf..266d455 100644 --- a/net-guardia/src/infrastructure/service_factory.rs +++ b/net-guardia/src/infrastructure/service_factory.rs @@ -185,14 +185,6 @@ impl ServiceFactory { Duration::from_secs(drift_window_secs), ))); - let app_services = Arc::new(AppServices::new( - app_config.clone(), - inference_config.clone(), - ml_manifest.clone(), - drift_detector.clone(), - ebpf_health.clone(), - )?); - // Create AtomicU8 enforce-level cache (Monitor=0, MlOnly=1, Enforce=2) let enforce_level_cache = Arc::new(AtomicU8::new({ use crate::infrastructure::enforce_mode_handler::enforce_mode_to_u8; @@ -200,8 +192,25 @@ impl ServiceFactory { enforce_mode_to_u8(&mode_str) })); - // Create CommunicationManager and register enforce-mode handler + // CommunicationManager and its event channels must exist before + // AppServices spins up the TrafficLogger: the writer thread can + // publish `flow_trace_stopped` audit events the moment it tries + // to open its first rotated file, and an unregistered channel + // would silently drop that evidence. let comm = Arc::new(CommunicationManager::new()); + comm.register_event_type::(); + comm.register_event_type::(); + comm.register_event_type::(); + + let app_services = Arc::new(AppServices::new( + app_config.clone(), + inference_config.clone(), + ml_manifest.clone(), + drift_detector.clone(), + ebpf_health.clone(), + comm.clone(), + )?); + let enforce_handler = Arc::new(EnforceModeHandler::new( db.clone() as Arc, comm.clone(), @@ -214,11 +223,6 @@ impl ServiceFactory { .query::() .build(); - // Register event type channels - comm.register_event_type::(); - comm.register_event_type::(); - comm.register_event_type::(); - // Seed default SOAR playbooks if empty (db.as_ref() as &dyn SoarRepo).seed_default_playbooks()?;