DaLaw2/NetGuardia
1
0
Fork 0
mirror of https://github.com/DaLaw2/NetGuardia.git synced 2026-08-24 14:10:28 +09:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Remove CommunicationManager

Browse Source
This commit is contained in:
DaLaw2 2026-04-25 13:07:57 +08:00
parent 94d366bcbe
commit 301d6ac1b8
32 changed files with 390 additions and 718 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
macros/src/config.rs
View File

@ -26,7 +26,9 @@ impl Parse for StructAttr {
input.parse::<Token![,]>()?;
}
}
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<String>,
) -> Vec<MappedSetting> {
fn parse_struct_mapped_settings(input: &mut ItemStruct, default_section: &Option<String>) -> Vec<MappedSetting> {
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<String>,
) -> Option<ConfigField> {
let idx = field
.attrs
.iter()
.position(|a| a.path().is_ident("setting"))?;
fn parse_field(field: &mut syn::Field, default_section: &Option<String>) -> Option<ConfigField> {
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<String, Vec<MappedSetting>> = 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 {

16
net-guardia/src/adapter/http/ml.rs
View File

@ -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<Inference>) -> impl Responder {
/// the upload path's `model_swap` so both swap-in and revert are auditable.
async fn delete_current_model(
inference: web::Data<Inference>,
comm: web::Data<CommunicationManager>,
audit_tx: web::Data<broadcast::Sender<AuditEvent>>,
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,

20
net-guardia/src/adapter/http/model_upload.rs
View File

@ -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<ArcSwap<AppConfig>>,
inference: web::Data<Inference>,
comm: web::Data<CommunicationManager>,
audit_tx: web::Data<broadcast::Sender<AuditEvent>>,
promote_lock: web::Data<PromoteGate>,
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<AuditEvent>,
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,

12
net-guardia/src/adapter/http/system.rs
View File

@ -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<CommunicationManager>) -> impl Responder {
match comm.send_query(GetEnforceModeQuery).await {
async fn get_enforce_mode(handler: web::Data<EnforceModeHandler>) -> 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<CommunicationManager>) -> impl Respond
async fn set_enforce_mode(
body: web::Json<EnforceModeRequest>,
comm: web::Data<CommunicationManager>,
handler: web::Data<EnforceModeHandler>,
) -> 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()})),
}

19
net-guardia/src/adapter/websocket/fusion_websocket.rs
View File

@ -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<CommunicationManager>,
threat_tx: web::Data<broadcast::Sender<ThreatDetectedEvent>>,
) -> Result<HttpResponse> {
let (response, session, msg_stream) = handle(&req, body)?;
let broadcast_rx = match comm.subscribe_event::<ThreatDetectedEvent>() {
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;

7
net-guardia/src/adapter/websocket/routes.rs
View File

@ -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<CommunicationManager>,
threat_tx: web::Data<broadcast::Sender<ThreatDetectedEvent>>,
query: web::Query<WsQuery>,
jwt: web::Data<JwtService>,
) -> 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)}))

25
net-guardia/src/core/detection/orchestrator.rs
View File

@ -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<DetectionEvent>,
comm: Arc<CommunicationManager>,
threat_tx: broadcast::Sender<ThreatDetectedEvent>,
audit_tx: broadcast::Sender<AuditEvent>,
geoip: Option<Arc<dyn GeoLookup>>,
metrics: Arc<FusionMetrics>,
// Enrichment state
@ -70,7 +70,8 @@ impl DetectionOrchestrator {
pub fn new(
app_config: &Arc<ArcSwap<AppConfig>>,
rx: mpsc::Receiver<DetectionEvent>,
comm: Arc<CommunicationManager>,
threat_tx: broadcast::Sender<ThreatDetectedEvent>,
audit_tx: broadcast::Sender<AuditEvent>,
geoip: Option<Arc<dyn GeoLookup>>,
metrics: Arc<FusionMetrics>,
) -> Self {
@ -79,7 +80,8 @@ impl DetectionOrchestrator {
let max_dedup = NonZero::new(fusion.max_dedup_entries.max(1)).unwrap_or(NonZero::<usize>::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<serde_json::Value> = 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.

36
net-guardia/src/core/inference/traffic_logger.rs
View File

@ -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<String>,
policy: RotationPolicy,
channel_capacity: usize,
comm: Option<Arc<CommunicationManager>>,
audit_tx: Option<broadcast::Sender<AuditEvent>>,
) -> Result<Self, io::Error> {
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<String>,
policy: RotationPolicy,
comm: Option<Arc<CommunicationManager>>,
audit_tx: Option<broadcast::Sender<AuditEvent>>,
) {
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<CommunicationManager>>, reason: &str, directory: &Path) {
let Some(c) = comm else {
fn emit_flow_trace_stop_audit(audit_tx: Option<&broadcast::Sender<AuditEvent>>, 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<CommunicationManager>>, 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::<AuditEvent>();
let mut rx = comm.subscribe_event::<AuditEvent>().unwrap();
async fn stop_audit_reaches_subscriber_when_sender_provided() {
let (tx, mut rx) = broadcast::channel::<AuditEvent>(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"));
}

15
net-guardia/src/core/response/engine.rs
View File

@ -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<Self>, comm: Arc<CommunicationManager>) -> Result<(), Error> {
let rx = comm.subscribe_event::<ThreatDetectedEvent>().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<Self>, threat_rx: broadcast::Receiver<ThreatDetectedEvent>) {
tokio::spawn(async move {
Self::event_loop(self, rx).await;
Self::event_loop(self, threat_rx).await;
});
Ok(())
}
async fn event_loop(self: Arc<Self>, mut rx: broadcast::Receiver<ThreatDetectedEvent>) {

3
net-guardia/src/domain/common/config/constants.rs
View File

@ -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;

3
net-guardia/src/domain/common/error/system.rs
View File

@ -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,

2
net-guardia/src/domain/common/event.rs
View File

@ -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,

3
net-guardia/src/domain/common/log/audit.rs
View File

@ -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,
}
}

3
net-guardia/src/domain/common/log/http.rs
View File

@ -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,
}
}

6
net-guardia/src/domain/detection/log.rs
View File

@ -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,
}

7
net-guardia/src/infrastructure/app_services.rs
View File

@ -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<ModelManifest>,
drift_detector: DriftDetectorHandle,
ebpf_health: Arc<ArcSwap<EbpfHealth>>,
comm: Arc<CommunicationManager>,
audit_tx: broadcast::Sender<AuditEvent>,
) -> Result<Self, Error> {
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));

25
net-guardia/src/infrastructure/audit_logger.rs
View File

@ -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<Self>, comm: &CommunicationManager) {
// Subscribe to AuditEvent
if let Ok(mut rx) = comm.subscribe_event::<AuditEvent>() {
/// Start background tasks that persist audit and drift events to DB + structured logs.
pub fn start(
self: Arc<Self>,
audit_rx: broadcast::Receiver<AuditEvent>,
drift_rx: broadcast::Receiver<DriftDetectedEvent>,
) {
// 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::<DriftDetectedEvent>() {
// 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);
}
}

373
net-guardia/src/infrastructure/communication_manager.rs
View File

@ -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<E: Event> {
pub sender: broadcast::Sender<E>,
}
impl<E: Event + 'static> EventBroadcaster for TypedEventBroadcaster<E> {
fn subscribe_typed(&self) -> Box<dyn Any + Send> {
Box::new(self.sender.subscribe())
}
fn broadcast_event(&self, event: Box<dyn Any + Send>) -> Result<(), Error> {
let typed_event = *event.downcast::<E>().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<TypeId, CommandHandlerFn>,
query_handlers: DashMap<TypeId, QueryHandlerFn>,
event_broadcasters: DashMap<TypeId, Box<dyn EventBroadcaster>>,
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<S: Send + Sync + 'static>(self: Arc<Self>, service: Arc<S>) -> ServiceRegistrar<S> {
ServiceRegistrar::new(service, self)
}
pub fn register_command_handler<C: Command + 'static>(&self, handler: Arc<dyn CommandHandler<C> + Send + Sync>) {
let type_id = TypeId::of::<C>();
let boxed_handler: CommandHandlerFn = Box::new(move |command: Box<dyn Any + Send>| {
let handler = handler.clone();
Box::pin(async move {
let command = *command.downcast::<C>().map_err(|_| MiscError::TypeMismatch)?;
handler.handle_command(command).await
}) as CommandFuture
});
self.command_handlers.insert(type_id, boxed_handler);
}
pub async fn send_command<C: Command + 'static>(&self, command: C) -> Result<(), Error> {
let type_id = TypeId::of::<C>();
if let Some(handler) = self.command_handlers.get(&type_id) {
handler(Box::new(command)).await
} else {
Err(MiscError::HandlerNotFound)?
}
}
pub fn register_query_handler<Q: Query + 'static>(&self, handler: Arc<dyn QueryHandler<Q> + Send + Sync>) {
let type_id = TypeId::of::<Q>();
let boxed_handler: QueryHandlerFn = Box::new(move |query: Box<dyn Any + Send>| {
let handler = handler.clone();
Box::pin(async move {
let query = *query.downcast::<Q>().map_err(|_| MiscError::TypeMismatch)?;
let response = handler.handle_query(query).await?;
Ok(Box::new(response) as Box<dyn Any + Send>)
}) as QueryFuture
});
self.query_handlers.insert(type_id, boxed_handler);
}
pub async fn send_query<Q: Query + 'static>(&self, query: Q) -> Result<Q::Response, Error> {
let type_id = TypeId::of::<Q>();
if let Some(handler) = self.query_handlers.get(&type_id) {
let response = handler(Box::new(query)).await?;
Ok(*response
.downcast::<Q::Response>()
.map_err(|_| MiscError::TypeMismatch)?)
} else {
Err(MiscError::HandlerNotFound)?
}
}
pub fn register_event_type<E: Event + 'static>(&self) {
let type_id = TypeId::of::<E>();
let (tx, _) = broadcast::channel::<E>(self.channel_capacity);
let broadcaster = TypedEventBroadcaster { sender: tx };
self.event_broadcasters.insert(type_id, Box::new(broadcaster));
}
pub fn subscribe_event<E: Event + 'static>(&self) -> Result<broadcast::Receiver<E>, Error> {
let type_id = TypeId::of::<E>();
let broadcaster = self
.event_broadcasters
.get(&type_id)
.ok_or(MiscError::TypeNotRegistered)?;
let receiver_box = broadcaster.subscribe_typed();
let receiver = *receiver_box
.downcast::<broadcast::Receiver<E>>()
.map_err(|_| MiscError::TypeMismatch)?;
Ok(receiver)
}
pub async fn publish_event<E: Event + 'static>(&self, event: E) -> Result<(), Error> {
let type_id = TypeId::of::<E>();
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<E: Event + 'static>(&self, event: E) -> Result<(), Error> {
let type_id = TypeId::of::<E>();
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<S> {
service: Arc<S>,
comm: Arc<CommunicationManager>,
}
impl<S: Send + Sync + 'static> ServiceRegistrar<S> {
fn new(service: Arc<S>, comm: Arc<CommunicationManager>) -> Self {
Self { service, comm }
}
pub fn command<C: Command + 'static>(self) -> Self
where
S: CommandHandler<C>,
{
let handler: Arc<dyn CommandHandler<C> + Send + Sync> = self.service.clone();
self.comm.register_command_handler::<C>(handler);
self
}
pub fn query<Q: Query + 'static>(self) -> Self
where
S: QueryHandler<Q>,
{
let handler: Arc<dyn QueryHandler<Q> + Send + Sync> = self.service.clone();
self.comm.register_query_handler::<Q>(handler);
self
}
pub fn build(self) -> Arc<CommunicationManager> {
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<Mutex<Vec<String>>>,
}
#[async_trait]
impl CommandHandler<TestCommand> 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<TestQuery> for TestQueryHandler {
async fn handle_query(&self, query: TestQuery) -> Result<i32, Error> {
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::<TestCommand>(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::<TestQuery>(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::<TestEvent>();
let mut receiver = comm.subscribe_event::<TestEvent>().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::<TestEvent>();
assert!(result.is_err());
}
#[tokio::test]
async fn publish_event_sync_delivers_to_subscriber() {
let comm = CommunicationManager::new(256);
comm.register_event_type::<TestEvent>();
let mut rx = comm.subscribe_event::<TestEvent>().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::<TestEvent>();
let mut rx1 = comm.subscribe_event::<TestEvent>().unwrap();
let mut rx2 = comm.subscribe_event::<TestEvent>().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::<TestCommand>().build();
comm.send_command(TestCommand {
value: "via_registrar".into(),
})
.await
.unwrap();
let msgs = received.lock().unwrap();
assert_eq!(msgs[0], "via_registrar");
}
}

121
net-guardia/src/infrastructure/enforce_mode_handler.rs
View File

@ -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<dyn AppRepo>,
comm: Arc<CommunicationManager>,
audit_tx: broadcast::Sender<AuditEvent>,
/// Shared AtomicU8 cache: Monitor=0, MlOnly=1, Enforce=2.
enforce_cache: Arc<AtomicU8>,
}
impl EnforceModeHandler {
pub fn new(db: Arc<dyn AppRepo>, comm: Arc<CommunicationManager>, enforce_cache: Arc<AtomicU8>) -> Self {
pub fn new(db: Arc<dyn AppRepo>, audit_tx: broadcast::Sender<AuditEvent>, enforce_cache: Arc<AtomicU8>) -> Self {
Self {
db,
comm,
audit_tx,
enforce_cache,
}
}
}
#[async_trait]
impl CommandHandler<ChangeEnforceModeCommand> 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<GetEnforceModeQuery> for EnforceModeHandler {
async fn handle_query(&self, _query: GetEnforceModeQuery) -> Result<String, Error> {
pub fn get_mode(&self) -> Result<String, Error> {
match self.db.get_setting("enforce_mode")? {
Some(mode) => Ok(mode),
None => Ok("monitor".to_string()),
@ -77,73 +62,53 @@ impl QueryHandler<GetEnforceModeQuery> 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<EnforceModeHandler>, Arc<CommunicationManager>) {
fn test_handler() -> EnforceModeHandler {
let db = Arc::new(Database::new(":memory:").unwrap()) as Arc<dyn AppRepo>;
let cache = Arc::new(AtomicU8::new(0));
let comm = Arc::new(CommunicationManager::new(256));
comm.register_event_type::<AuditEvent>();
let handler = Arc::new(EnforceModeHandler::new(db, comm.clone(), cache));
let _ = comm
.clone()
.with_service(handler.clone())
.command::<ChangeEnforceModeCommand>()
.query::<GetEnforceModeQuery>()
.build();
(handler, comm)
let (audit_tx, _rx) = broadcast::channel::<AuditEvent>(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);
}
}

16
net-guardia/src/infrastructure/http_server.rs
View File

@ -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<Database>,
pub secret_store: Arc<SecretStore>,
pub jwt_service: Arc<JwtService>,
pub comm: Arc<CommunicationManager>,
pub threat_tx: broadcast::Sender<ThreatDetectedEvent>,
pub audit_tx: broadcast::Sender<AuditEvent>,
pub enforce_handler: Arc<EnforceModeHandler>,
pub setup_complete: SetupCompleteFlag,
pub ready: ReadyFlag,
pub readiness_state: Arc<ReadinessState>,
@ -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()))

200
net-guardia/src/infrastructure/logging.rs Normal file
View File

@ -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<Box<dyn FilterControl>> = 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<Vec<String>> = 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<String> = 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::<Directive>() {
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<String, String> {
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::<Directive>() {
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 `<select>` that only knows five options.
pub fn current_level() -> String {
FILTER_HANDLE
.get()
.map(|h| extract_main_level(&h.current_filter()))
.unwrap_or_else(|| "unknown".to_string())
}
}
impl<L> FilterControl for reload::Handle<EnvFilter, L> {
fn reload_filter(&self, filter: EnvFilter) -> Result<(), String> {
self.reload(filter).map_err(|e| e.to_string())
}
fn current_filter(&self) -> String {
self.with_current(|f| f.to_string())
.unwrap_or_else(|_| "unknown".to_string())
}
}
/// Strip per-target directives out of an EnvFilter string and return the
/// bare level directive in lowercase. Falls back to the raw string if no
/// bare directive is present.
fn extract_main_level(raw: &str) -> String {
raw.split(',')
.map(str::trim)
.find(|d| !d.is_empty() && !d.contains('='))
.unwrap_or(raw)
.to_lowercase()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn strips_per_target_directives() {
assert_eq!(extract_main_level("maxminddb=warn,debug"), "debug");
assert_eq!(extract_main_level("info,maxminddb=warn"), "info");
assert_eq!(extract_main_level("DEBUG"), "debug");
}
#[test]
fn falls_back_when_no_bare_level() {
// Only per-target directives → return lowercased raw so the UI at
// least shows *something* rather than silently misleading.
assert_eq!(extract_main_level("maxminddb=warn"), "maxminddb=warn");
}
}

1
net-guardia/src/infrastructure/mod.rs
View File

@ -1,7 +1,6 @@
pub mod app_services;
pub mod audit_logger;
pub mod cli;
pub mod communication_manager;
pub mod ebpf_preflight;
pub mod enforce_mode_handler;
pub mod geoip;

43
net-guardia/src/infrastructure/service_factory.rs
View File

@ -12,6 +12,7 @@ use aya::programs::{Xdp, XdpFlags};
use aya_log::EbpfLogger;
use common::define::pipeline::*;
use macros::log;
use tokio::sync::broadcast;
use crate::adapter::access_control::AccessControlAdapter;
use crate::adapter::ebpf::EbpfServices;
@ -29,6 +30,7 @@ use crate::core::response::engine::SoarEngine;
use crate::core::response::playbook_service::PlaybookService;
use crate::core::response::scheduler::TtlScheduler;
use crate::domain::common::config::AppConfig;
use crate::domain::common::config::constants::EVENT_CHANNEL_CAPACITY;
use crate::domain::common::error::Error;
use crate::domain::common::error::misc::MiscError;
use crate::domain::common::event::{AuditEvent, DriftDetectedEvent, ThreatDetectedEvent};
@ -44,14 +46,11 @@ use crate::domain::detection::log::MLLog;
use crate::domain::detection::manifest::ModelManifest;
use crate::domain::detection::ml_inference_config::MLInferenceConfig;
use crate::infrastructure::app_services::AppServices;
use crate::infrastructure::communication_manager::CommunicationManager;
use crate::infrastructure::ebpf_preflight;
use crate::infrastructure::enforce_mode_handler::EnforceModeHandler;
use crate::infrastructure::geoip::GeoIpService;
use crate::infrastructure::secret_store::SecretStore;
use crate::infrastructure::suricata_manager::SuricataManager;
use crate::interface::communication::command_types::ChangeEnforceModeCommand;
use crate::interface::communication::query_types::GetEnforceModeQuery;
use crate::interface::port::access_control::AccessControlPort;
use crate::interface::port::access_control_admin::AccessControlAdminPort;
use crate::interface::port::app_repo::AppRepo;
@ -73,7 +72,10 @@ pub struct AppState {
pub database: Arc<Database>,
pub secret_store: Arc<SecretStore>,
pub jwt_service: Arc<JwtService>,
pub comm: Arc<CommunicationManager>,
pub threat_tx: broadcast::Sender<ThreatDetectedEvent>,
pub audit_tx: broadcast::Sender<AuditEvent>,
pub drift_tx: broadcast::Sender<DriftDetectedEvent>,
pub enforce_handler: Arc<EnforceModeHandler>,
pub soar_engine: Arc<SoarEngine>,
pub ttl_scheduler: TtlScheduler,
pub report_scheduler: ReportScheduler,
@ -199,17 +201,13 @@ impl ServiceFactory {
enforce_mode_to_u8(&mode_str)
}));
// 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(
app_config.load().observability.default_event_channel_capacity,
));
comm.register_event_type::<ThreatDetectedEvent>();
comm.register_event_type::<DriftDetectedEvent>();
comm.register_event_type::<AuditEvent>();
// Named broadcast channels replace the old TypeId-based CommunicationManager.
// They 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.
let (threat_tx, _) = broadcast::channel::<ThreatDetectedEvent>(EVENT_CHANNEL_CAPACITY);
let (audit_tx, _) = broadcast::channel::<AuditEvent>(EVENT_CHANNEL_CAPACITY);
let (drift_tx, _) = broadcast::channel::<DriftDetectedEvent>(EVENT_CHANNEL_CAPACITY);
let app_services = Arc::new(AppServices::new(
app_config.clone(),
@ -217,20 +215,14 @@ impl ServiceFactory {
ml_manifest.clone(),
drift_detector.clone(),
ebpf_health.clone(),
comm.clone(),
audit_tx.clone(),
)?);
let enforce_handler = Arc::new(EnforceModeHandler::new(
db.clone() as Arc<dyn AppRepo>,
comm.clone(),
audit_tx.clone(),
enforce_level_cache.clone(),
));
let _ = comm
.clone()
.with_service(enforce_handler)
.command::<ChangeEnforceModeCommand>()
.query::<GetEnforceModeQuery>()
.build();
// Seed default SOAR playbooks if empty
(db.as_ref() as &dyn SoarRepo).seed_default_playbooks()?;
@ -340,7 +332,10 @@ impl ServiceFactory {
database: db,
secret_store,
jwt_service,
comm,
threat_tx,
audit_tx,
drift_tx,
enforce_handler,
soar_engine,
ttl_scheduler,
report_scheduler,

36
net-guardia/src/infrastructure/system.rs
View File

@ -9,7 +9,7 @@ use aya::maps::{MapData, ProgramArray};
use macros::log;
use sd_notify::NotifyState;
use tokio::signal::ctrl_c;
use tokio::sync::broadcast::{Receiver, error::RecvError};
use tokio::sync::broadcast::{self, Receiver, error::RecvError};
use tokio::sync::mpsc::{self, Sender};
use tokio::sync::oneshot;
use tokio::time::{interval, sleep};
@ -38,7 +38,8 @@ use crate::domain::common::config::AppConfig;
use crate::domain::common::config::constants::{MODELS_DIR, STAGING_SUBDIR};
use crate::domain::common::error::Error;
use crate::domain::common::error::system::SystemError;
use crate::domain::common::event::{DetectionEvent, DetectionSource, DriftDetectedEvent};
use crate::domain::common::event::AuditEvent;
use crate::domain::common::event::{DetectionEvent, DetectionSource, DriftDetectedEvent, ThreatDetectedEvent};
use crate::domain::common::log::system::SystemLog;
use crate::domain::common::system::health::EbpfHealth;
use crate::domain::common::system::readiness::ReadinessState;
@ -48,7 +49,7 @@ use crate::domain::detection::ml_detection::AlertMessage;
use crate::domain::detection::ml_inference_config::MLInferenceConfig;
use crate::infrastructure::app_services::AppServices;
use crate::infrastructure::audit_logger::AuditLogger;
use crate::infrastructure::communication_manager::CommunicationManager;
use crate::infrastructure::enforce_mode_handler::EnforceModeHandler;
use crate::infrastructure::http_server::{self, HttpServerParams};
use crate::infrastructure::logger::Logger;
use crate::infrastructure::secret_store::SecretStore;
@ -89,7 +90,10 @@ pub struct System {
pub database: Arc<Database>,
pub secret_store: Arc<SecretStore>,
pub jwt_service: Arc<JwtService>,
pub comm: Arc<CommunicationManager>,
pub threat_tx: broadcast::Sender<ThreatDetectedEvent>,
pub audit_tx: broadcast::Sender<AuditEvent>,
pub drift_tx: broadcast::Sender<DriftDetectedEvent>,
pub enforce_handler: Arc<EnforceModeHandler>,
pub soar_engine: Arc<SoarEngine>,
pub ttl_scheduler: Option<TtlScheduler>,
pub report_scheduler: Option<ReportScheduler>,
@ -121,7 +125,10 @@ impl System {
database: state.database,
secret_store: state.secret_store,
jwt_service: state.jwt_service,
comm: state.comm,
threat_tx: state.threat_tx,
audit_tx: state.audit_tx,
drift_tx: state.drift_tx,
enforce_handler: state.enforce_handler,
soar_engine: state.soar_engine,
ttl_scheduler: Some(state.ttl_scheduler),
report_scheduler: Some(state.report_scheduler),
@ -215,7 +222,7 @@ impl System {
// Start SOAR engine
self.soar_engine.recover_active_blocks().await?;
self.soar_engine.clone().start(self.comm.clone())?;
self.soar_engine.clone().start(self.threat_tx.subscribe());
// Start TTL scheduler
if let Some(ttl) = self.ttl_scheduler.take() {
@ -229,7 +236,7 @@ impl System {
// Start audit logger (subscribe to AuditEvent + DriftDetectedEvent, persist to DB)
let audit_logger = Arc::new(AuditLogger::new(self.database.clone() as Arc<dyn AuditRepo>));
audit_logger.start(&self.comm);
audit_logger.start(self.audit_tx.subscribe(), self.drift_tx.subscribe());
// Start stats aggregator (writes weekly_* settings for Report engine)
let stats_aggregator = StatsAggregator::new(
@ -241,9 +248,9 @@ impl System {
// Start drift detection background task
{
let drift_detector = self.drift_detector.clone();
let comm_drift = self.comm.clone();
let drift_tx = self.drift_tx.clone();
tokio::spawn(async move {
Self::run_drift_monitor(drift_detector, comm_drift).await;
Self::run_drift_monitor(drift_detector, drift_tx).await;
});
}
@ -252,7 +259,8 @@ impl System {
let orchestrator = DetectionOrchestrator::new(
&self.app_config,
detection_rx,
self.comm.clone(),
self.threat_tx.clone(),
self.audit_tx.clone(),
self.geoip.clone(),
self.app_services.fusion_metrics.clone(),
);
@ -322,7 +330,9 @@ impl System {
db: self.database.clone(),
secret_store: self.secret_store.clone(),
jwt_service: self.jwt_service.clone(),
comm: self.comm.clone(),
threat_tx: self.threat_tx.clone(),
audit_tx: self.audit_tx.clone(),
enforce_handler: self.enforce_handler.clone(),
setup_complete: setup_flag,
ready: ready_flag,
readiness_state,
@ -406,7 +416,7 @@ impl System {
}
/// Periodically check the drift detector and publish DriftDetectedEvent when drift is found.
async fn run_drift_monitor(drift_detector: DriftDetectorHandle, comm: Arc<CommunicationManager>) {
async fn run_drift_monitor(drift_detector: DriftDetectorHandle, drift_tx: broadcast::Sender<DriftDetectedEvent>) {
let mut interval = interval(Duration::from_secs(60));
loop {
interval.tick().await;
@ -420,7 +430,7 @@ impl System {
drifted_features: report.drifted_features,
max_deviation: report.max_deviation,
};
if let Err(e) = comm.publish_event(event).await {
if let Err(e) = drift_tx.send(event) {
log!(SystemError::DriftEventPublishFailed(e));
}
}

18
net-guardia/src/interface/communication/command.rs
View File

@ -1,18 +0,0 @@
use std::any::Any;
use std::future::Future;
use std::pin::Pin;
use async_trait::async_trait;
use crate::domain::common::error::Error;
use crate::interface::communication::message::Message;
pub type CommandFuture = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
pub type CommandHandlerFn = Box<dyn Fn(Box<dyn Any + Send>) -> CommandFuture + Send + Sync>;
pub trait Command: Message<Response = ()> {}
#[async_trait]
pub trait CommandHandler<C: Command> {
async fn handle_command(&self, command: C) -> Result<(), Error>;
}

13
net-guardia/src/interface/communication/command_types.rs
View File

@ -1,13 +0,0 @@
use crate::interface::communication::command::Command;
use crate::interface::communication::message::Message;
// ── System Commands ──────────────────────────────────────────────────
pub struct ChangeEnforceModeCommand {
pub mode: String,
}
impl Message for ChangeEnforceModeCommand {
type Response = ();
}
impl Command for ChangeEnforceModeCommand {}

9
net-guardia/src/interface/communication/event.rs
View File

@ -1,10 +1 @@
use std::any::Any;
use crate::domain::common::error::Error;
pub trait Event: Send + Clone + 'static {}
pub trait EventBroadcaster: Send + Sync {
fn subscribe_typed(&self) -> Box<dyn Any + Send>;
fn broadcast_event(&self, event: Box<dyn Any + Send>) -> Result<(), Error>;
}

3
net-guardia/src/interface/communication/message.rs
View File

@ -1,3 +0,0 @@
pub trait Message: Send + 'static {
type Response: Send + 'static;
}

5
net-guardia/src/interface/communication/mod.rs
View File

@ -1,7 +1,2 @@
pub mod command;
pub mod command_types;
pub mod event;
pub mod event_types;
pub mod message;
pub mod query;
pub mod query_types;

18
net-guardia/src/interface/communication/query.rs
View File

@ -1,18 +0,0 @@
use std::any::Any;
use std::future::Future;
use std::pin::Pin;
use async_trait::async_trait;
use crate::domain::common::error::Error;
use crate::interface::communication::message::Message;
pub type QueryFuture = Pin<Box<dyn Future<Output = Result<Box<dyn Any + Send>, Error>> + Send + 'static>>;
pub type QueryHandlerFn = Box<dyn Fn(Box<dyn Any + Send>) -> QueryFuture + Send + Sync>;
pub trait Query: Message {}
#[async_trait]
pub trait QueryHandler<Q: Query> {
async fn handle_query(&self, query: Q) -> Result<Q::Response, Error>;
}

11
net-guardia/src/interface/communication/query_types.rs
View File

@ -1,11 +0,0 @@
use crate::interface::communication::message::Message;
use crate::interface::communication::query::Query;
// ── System Queries ───────────────────────────────────────────────────
pub struct GetEnforceModeQuery;
impl Message for GetEnforceModeQuery {
type Response = String;
}
impl Query for GetEnforceModeQuery {}

6
net-guardia/src/main.rs
View File

@ -14,9 +14,6 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use clap::Parser;
use domain::common::error::Error;
use domain::common::error::system::SystemError;
use domain::common::log::system::SystemLog;
use macros::log;
use sd_notify::NotifyState;
use tokio::time::sleep;
@ -25,6 +22,9 @@ use tokio::{signal, time};
use crate::adapter::persistence::Database;
use crate::core::identity::jwt::JwtService;
use crate::domain::common::config::observability::ObservabilityConfig;
use crate::domain::common::error::Error;
use crate::domain::common::error::system::SystemError;
use crate::domain::common::log::system::SystemLog;
use crate::domain::identity::password;
use crate::infrastructure::cli::{Cli, handle_subcommand};
use crate::infrastructure::http_server;