Update Cargo.lock after adding network-types to egress-ebpf

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138PxtKH73hqxv7h1oaoSdS

Cargo.lock

@@ -691,6 +691,7 @@ dependencies = [
  "aya-ebpf",
  "aya-log-ebpf",
  "common",
+ "network-types",
  "which",
 ]
 

common/src/model/event.rs

@@ -193,4 +193,15 @@ impl TcpFlags {
             cwr: (flags & 0x80) != 0,
         }
     }
+
+    pub fn to_byte(&self) -> u8 {
+        ((self.cwr as u8) << 7)
+            | ((self.ece as u8) << 6)
+            | ((self.urg as u8) << 5)
+            | ((self.ack as u8) << 4)
+            | ((self.psh as u8) << 3)
+            | ((self.rst as u8) << 2)
+            | ((self.syn as u8) << 1)
+            | (self.fin as u8)
+    }
 }

common/src/model/mod.rs

@@ -5,3 +5,4 @@ pub mod ip_address;
 pub mod packet;
 pub mod placeholder;
 pub mod pseudo_header;
+pub mod tcp_anomaly;

common/src/model/tcp_anomaly.rs

@@ -0,0 +1,25 @@
+pub const ANOMALY_SYN_FIN: u8 = 0;
+pub const ANOMALY_NULL_SCAN: u8 = 1;
+pub const ANOMALY_XMAS_SCAN: u8 = 2;
+pub const ANOMALY_RST_SYN: u8 = 3;
+
+pub const DIRECTION_INGRESS: u8 = 0;
+pub const DIRECTION_EGRESS: u8 = 1;
+
+/// TCP flag anomaly event written to the ring buffer by XDP programs.
+///
+/// align_of::<TcpAnomalyEvent>() == 8, satisfying the aya ring buffer constraint
+/// that 8 % align_of::<T>() == 0.
+#[repr(C)]
+#[derive(Clone, Copy)]
+pub struct TcpAnomalyEvent {
+    pub timestamp_ns: u64,
+    pub src_ip: u32,
+    pub dst_ip: u32,
+    pub src_port: u16,
+    pub dst_port: u16,
+    pub flags: u8,
+    pub anomaly_type: u8,
+    pub direction: u8,
+    pub _pad: u8,
+}

egress-ebpf/Cargo.toml

@@ -8,6 +8,7 @@ common = { path = "../common", features = ["kernel"] }
 
 aya-ebpf = { workspace = true }
 aya-log-ebpf = { workspace = true }
+network-types = { workspace = true }
 
 [build-dependencies]
 which = "8.0.0"

egress-ebpf/src/action/anomaly.rs

@@ -0,0 +1,50 @@
+use aya_ebpf::macros::map;
+use aya_ebpf::maps::RingBuf;
+use common::model::event::IPv4Event;
+use common::model::tcp_anomaly::{
+    ANOMALY_NULL_SCAN, ANOMALY_RST_SYN, ANOMALY_SYN_FIN, ANOMALY_XMAS_SCAN, DIRECTION_EGRESS,
+    TcpAnomalyEvent,
+};
+use network_types::ip::IpProto;
+
+#[map]
+pub static TCP_ANOMALY_EVENTS: RingBuf = RingBuf::with_byte_size(1 << 18, 0);
+
+/// Returns true if a TCP flag anomaly was detected (and emitted to the ring buffer).
+#[inline(always)]
+pub fn ipv4_check_and_emit(event: &IPv4Event) -> bool {
+    if event.protocol != IpProto::Tcp {
+        return false;
+    }
+    let flags_byte = event.tcp_flags.to_byte();
+    let anomaly_type = if flags_byte == 0 {
+        ANOMALY_NULL_SCAN
+    } else if flags_byte & 0x03 == 0x03 {
+        ANOMALY_SYN_FIN
+    } else if flags_byte & 0x29 == 0x29 {
+        ANOMALY_XMAS_SCAN
+    } else if flags_byte & 0x06 == 0x06 {
+        ANOMALY_RST_SYN
+    } else {
+        return false;
+    };
+
+    let ev = TcpAnomalyEvent {
+        timestamp_ns: event.timestamp_us.wrapping_mul(1000),
+        src_ip: event.src_ip,
+        dst_ip: event.dst_ip,
+        src_port: event.src_port,
+        dst_port: event.dst_port,
+        flags: flags_byte,
+        anomaly_type,
+        direction: DIRECTION_EGRESS,
+        _pad: 0,
+    };
+
+    if let Some(mut entry) = TCP_ANOMALY_EVENTS.reserve::<TcpAnomalyEvent>(0) {
+        entry.write(ev);
+        entry.submit(0);
+    }
+
+    true
+}

egress-ebpf/src/action/mod.rs

@@ -1,2 +1,3 @@
 pub mod access_control;
+pub mod anomaly;
 pub mod statistics;

egress-ebpf/src/main.rs

@@ -2,7 +2,7 @@
 #![no_main]
 mod action;
 
-use action::{access_control, statistics};
+use action::{access_control, anomaly, statistics};
 use aya_ebpf::bindings::xdp_action;
 use aya_ebpf::macros::{map, xdp};
 use aya_ebpf::maps::{PerCpuArray, ProgramArray, XskMap};
@@ -68,6 +68,9 @@ unsafe fn try_access_control(ctx: &XdpContext) -> Result<u32, ()> {
             if access_control::ipv4_is_blacklisted(event) {
                 return Ok(xdp_action::XDP_DROP);
             }
+            if anomaly::ipv4_check_and_emit(event) {
+                return Ok(xdp_action::XDP_DROP);
+            }
         }
         Event::IPv6(event) => {
             if access_control::ipv6_is_whitelisted(event) {

ingress-ebpf/src/action/anomaly.rs

@@ -0,0 +1,50 @@
+use aya_ebpf::macros::map;
+use aya_ebpf::maps::RingBuf;
+use common::model::event::IPv4Event;
+use common::model::tcp_anomaly::{
+    ANOMALY_NULL_SCAN, ANOMALY_RST_SYN, ANOMALY_SYN_FIN, ANOMALY_XMAS_SCAN, DIRECTION_INGRESS,
+    TcpAnomalyEvent,
+};
+use network_types::ip::IpProto;
+
+#[map]
+pub static TCP_ANOMALY_EVENTS: RingBuf = RingBuf::with_byte_size(1 << 18, 0);
+
+/// Returns true if a TCP flag anomaly was detected (and emitted to the ring buffer).
+#[inline(always)]
+pub fn ipv4_check_and_emit(event: &IPv4Event) -> bool {
+    if event.protocol != IpProto::Tcp {
+        return false;
+    }
+    let flags_byte = event.tcp_flags.to_byte();
+    let anomaly_type = if flags_byte == 0 {
+        ANOMALY_NULL_SCAN
+    } else if flags_byte & 0x03 == 0x03 {
+        ANOMALY_SYN_FIN
+    } else if flags_byte & 0x29 == 0x29 {
+        ANOMALY_XMAS_SCAN
+    } else if flags_byte & 0x06 == 0x06 {
+        ANOMALY_RST_SYN
+    } else {
+        return false;
+    };
+
+    let ev = TcpAnomalyEvent {
+        timestamp_ns: event.timestamp_us.wrapping_mul(1000),
+        src_ip: event.src_ip,
+        dst_ip: event.dst_ip,
+        src_port: event.src_port,
+        dst_port: event.dst_port,
+        flags: flags_byte,
+        anomaly_type,
+        direction: DIRECTION_INGRESS,
+        _pad: 0,
+    };
+
+    if let Some(mut entry) = TCP_ANOMALY_EVENTS.reserve::<TcpAnomalyEvent>(0) {
+        entry.write(ev);
+        entry.submit(0);
+    }
+
+    true
+}

ingress-ebpf/src/action/mod.rs

@@ -1,2 +1,3 @@
 pub mod access_control;
+pub mod anomaly;
 pub mod statistics;

ingress-ebpf/src/main.rs

@@ -12,7 +12,7 @@ use common::define::program_array::ingress::*;
 use common::ebpf::parsing;
 use common::model::event::Event;
 
-use crate::action::{access_control, statistics};
+use crate::action::{access_control, anomaly, statistics};
 
 #[map]
 static PROGRAM_ARRAY: ProgramArray = ProgramArray::with_max_entries(8, 0);
@@ -70,6 +70,9 @@ unsafe fn try_access_control(ctx: &XdpContext) -> Result<u32, ()> {
             if access_control::ipv4_is_blacklisted(event) {
                 return Ok(xdp_action::XDP_DROP);
             }
+            if anomaly::ipv4_check_and_emit(event) {
+                return Ok(xdp_action::XDP_DROP);
+            }
         }
         Event::IPv6(event) => {
             if access_control::ipv6_is_whitelisted(event) {

mantis/src/core/ebpf/mod.rs

@@ -1,8 +1,10 @@
 pub mod access_control;
 pub mod statistics;
+pub mod tcp_anomaly_tracker;
 pub mod xsk_manager;
 
 use std::sync::Arc;
+use std::sync::Mutex;
 
 use aya::Ebpf;
 use crossbeam::queue::SegQueue;
@@ -11,8 +13,10 @@ use tokio::sync::oneshot;
 
 use crate::core::ebpf::access_control::AccessControl;
 use crate::core::ebpf::statistics::Statistics;
+use crate::core::ebpf::tcp_anomaly_tracker::TcpAnomalyTracker;
 use crate::core::ebpf::xsk_manager::XskManager;
 use crate::core::infrastructure::app_config::AppConfig;
+use crate::core::infrastructure::detection_alert::SharedDetectionAlert;
 use crate::detection::ml::engine::Engine;
 use crate::detection::suricata::SuricataEngine;
 use crate::model::error::Error;
@@ -22,6 +26,7 @@ pub struct EbpfServices {
     pub xsk_manager: Arc<XskManager>,
     pub access_control: Arc<AccessControl>,
     pub statistics: Arc<Statistics>,
+    tcp_anomaly_tracker: Mutex<Option<TcpAnomalyTracker>>,
     pub shutdowns: SegQueue<oneshot::Sender<()>>,
 }
 
@@ -30,10 +35,12 @@ impl EbpfServices {
         let xsk_manager = XskManager::new(app_config.clone(), ingress_ebpf, egress_ebpf)?;
         let access_control = AccessControl::new(ingress_ebpf, egress_ebpf)?;
         let statistics = Statistics::new(app_config.clone(), ingress_ebpf, egress_ebpf)?;
+        let tcp_anomaly_tracker = TcpAnomalyTracker::new(ingress_ebpf, egress_ebpf)?;
         let ebpf_services = Self {
             xsk_manager: Arc::new(xsk_manager),
             access_control: Arc::new(access_control),
             statistics: Arc::new(statistics),
+            tcp_anomaly_tracker: Mutex::new(Some(tcp_anomaly_tracker)),
             shutdowns: SegQueue::new(),
         };
         Ok(ebpf_services)
@@ -43,6 +50,7 @@ impl EbpfServices {
         self: Arc<Self>,
         ml_engine: Arc<Engine>,
         suricata_engine: Option<Arc<SuricataEngine>>,
+        detection_alert: SharedDetectionAlert,
     ) -> Result<(), Error> {
         let xsk_manager = self.xsk_manager.clone();
         let statistics = self.statistics.clone();
@@ -51,6 +59,12 @@ impl EbpfServices {
 
         let statistics_shutdown = statistics.run().await;
         self.shutdowns.push(statistics_shutdown);
+
+        if let Some(tracker) = self.tcp_anomaly_tracker.lock().unwrap().take() {
+            let anomaly_shutdown = tracker.run(detection_alert)?;
+            self.shutdowns.push(anomaly_shutdown);
+        }
+
         Ok(())
     }
 

mantis/src/core/ebpf/tcp_anomaly_tracker.rs

@@ -0,0 +1,81 @@
+use aya::maps::{MapData, RingBuf};
+use common::model::tcp_anomaly::TcpAnomalyEvent;
+use tokio::io::unix::AsyncFd;
+use tokio::sync::oneshot;
+
+use crate::core::infrastructure::detection_alert::SharedDetectionAlert;
+use crate::model::error::Error;
+use crate::model::error::ebpf::EbpfError;
+use crate::model::alert::UnifiedAlert;
+
+pub struct TcpAnomalyTracker {
+    ingress_ring_buf: RingBuf<MapData>,
+    egress_ring_buf: RingBuf<MapData>,
+}
+
+impl TcpAnomalyTracker {
+    pub fn new(ingress_ebpf: &mut aya::Ebpf, egress_ebpf: &mut aya::Ebpf) -> Result<Self, Error> {
+        let ingress_map = ingress_ebpf
+            .take_map("TCP_ANOMALY_EVENTS")
+            .ok_or(EbpfError::MapNotFound)?;
+        let egress_map = egress_ebpf
+            .take_map("TCP_ANOMALY_EVENTS")
+            .ok_or(EbpfError::MapNotFound)?;
+        let ingress_ring_buf =
+            RingBuf::try_from(ingress_map).map_err(EbpfError::MapOperationError)?;
+        let egress_ring_buf =
+            RingBuf::try_from(egress_map).map_err(EbpfError::MapOperationError)?;
+        Ok(Self { ingress_ring_buf, egress_ring_buf })
+    }
+
+    pub fn run(self, detection_alert: SharedDetectionAlert) -> Result<oneshot::Sender<()>, Error> {
+        let (tx, rx) = oneshot::channel::<()>();
+
+        let ingress_fd =
+            AsyncFd::new(self.ingress_ring_buf).map_err(EbpfError::MapOperationError)?;
+        let egress_fd =
+            AsyncFd::new(self.egress_ring_buf).map_err(EbpfError::MapOperationError)?;
+
+        tokio::spawn(async move {
+            let mut rx = rx;
+            let mut ingress_fd = ingress_fd;
+            let mut egress_fd = egress_fd;
+            loop {
+                tokio::select! {
+                    biased;
+                    _ = &mut rx => break,
+                    result = ingress_fd.readable_mut() => {
+                        if let Ok(mut guard) = result {
+                            let rb = guard.get_inner_mut();
+                            while let Some(item) = rb.next() {
+                                Self::process_event(&*item, &detection_alert);
+                            }
+                            guard.clear_ready();
+                        }
+                    }
+                    result = egress_fd.readable_mut() => {
+                        if let Ok(mut guard) = result {
+                            let rb = guard.get_inner_mut();
+                            while let Some(item) = rb.next() {
+                                Self::process_event(&*item, &detection_alert);
+                            }
+                            guard.clear_ready();
+                        }
+                    }
+                }
+            }
+        });
+
+        Ok(tx)
+    }
+
+    fn process_event(item: &[u8], detection_alert: &SharedDetectionAlert) {
+        if item.len() < core::mem::size_of::<TcpAnomalyEvent>() {
+            return;
+        }
+        let event = unsafe { &*(item.as_ptr() as *const TcpAnomalyEvent) };
+        if detection_alert.has_subscribers() {
+            detection_alert.broadcast(UnifiedAlert::from_tcp_anomaly(event));
+        }
+    }
+}

mantis/src/core/infrastructure/detection_alert.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use tokio::sync::broadcast;
 
-use crate::model::ml_detection::UnifiedAlert;
+use crate::model::alert::UnifiedAlert;
 
 pub struct DetectionAlert {
     broadcast_tx: broadcast::Sender<UnifiedAlert>,

mantis/src/core/system.rs

@@ -96,7 +96,11 @@ impl System {
         self.attach_ebpf()?;
 
         ebpf_services
-            .run(app_services.ml_engine.clone(), app_services.suricata_engine.clone())
+            .run(
+                app_services.ml_engine.clone(),
+                app_services.suricata_engine.clone(),
+                app_services.detection_alert.clone(),
+            )
             .await?;
         app_services.run().await?;
 

mantis/src/detection/fusion.rs

@@ -3,7 +3,8 @@ use std::sync::{Arc, Mutex};
 use std::time::Instant;
 
 use crate::core::infrastructure::detection_alert::DetectionAlert;
-use crate::model::ml_detection::{DetectionResult, UnifiedAlert};
+use crate::model::alert::UnifiedAlert;
+use crate::model::ml_detection::DetectionResult;
 use crate::model::rule_detection::RuleMatch;
 
 #[derive(Clone, Copy, PartialEq, Eq)]

mantis/src/model/alert.rs

@@ -0,0 +1,159 @@
+use common::model::tcp_anomaly::{
+    ANOMALY_NULL_SCAN, ANOMALY_RST_SYN, ANOMALY_SYN_FIN, ANOMALY_XMAS_SCAN, DIRECTION_INGRESS,
+    TcpAnomalyEvent,
+};
+use serde::Serialize;
+
+use crate::model::ml_detection::DetectionResult;
+use crate::model::rule_detection::RuleMatch;
+use crate::utils::packet_parser::format_ipv4;
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum AlertSource {
+    Ml,
+    Rule,
+    Fusion,
+    TcpAnomaly,
+}
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum AlertSeverity {
+    /// Single-source detection: either ML or Rule fired alone.
+    High,
+    /// Both ML and Rule agreed on the same flow within the fusion window.
+    Critical,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct UnifiedAlert {
+    pub timestamp: u64,
+    pub flow_key: String,
+    pub src_ip: String,
+    pub dst_ip: String,
+    pub src_port: u16,
+    pub dst_port: u16,
+    pub protocol: u8,
+    pub source: AlertSource,
+    pub severity: AlertSeverity,
+    pub is_attack: bool,
+    pub attack_type: Option<String>,
+    pub confidence: f32,
+    pub ae_score: f32,
+    pub rule_sid: Option<u32>,
+    pub rule_msg: Option<String>,
+}
+
+fn now_secs() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("system time is after UNIX_EPOCH")
+        .as_secs()
+}
+
+fn parse_ip_port(s: &str) -> (String, u16) {
+    match s.rfind(':') {
+        Some(pos) => {
+            let port = s[pos + 1..].parse::<u16>().unwrap_or(0);
+            (s[..pos].to_string(), port)
+        }
+        None => (s.to_string(), 0),
+    }
+}
+
+impl UnifiedAlert {
+    pub fn from_ml(result: &DetectionResult) -> Self {
+        Self {
+            timestamp: now_secs(),
+            flow_key: result.flow_key.clone(),
+            src_ip: result.flow_key_raw.src_ip.to_string(),
+            dst_ip: result.flow_key_raw.dst_ip.to_string(),
+            src_port: result.flow_key_raw.src_port,
+            dst_port: result.flow_key_raw.dst_port,
+            protocol: result.flow_key_raw.protocol,
+            source: AlertSource::Ml,
+            severity: AlertSeverity::High,
+            is_attack: result.is_attack,
+            attack_type: result.attack_type.clone(),
+            confidence: result.confidence,
+            ae_score: result.ae_score,
+            rule_sid: None,
+            rule_msg: None,
+        }
+    }
+
+    pub fn from_rule(m: &RuleMatch) -> Self {
+        let (src_ip, src_port) = parse_ip_port(&m.src);
+        let (dst_ip, dst_port) = parse_ip_port(&m.dst);
+        let flow_key = format!("{}->{}", m.src, m.dst);
+        Self {
+            timestamp: now_secs(),
+            flow_key,
+            src_ip,
+            dst_ip,
+            src_port,
+            dst_port,
+            protocol: 6,
+            source: AlertSource::Rule,
+            severity: AlertSeverity::High,
+            is_attack: true,
+            attack_type: None,
+            confidence: 1.0,
+            ae_score: 0.0,
+            rule_sid: Some(m.sid),
+            rule_msg: Some(m.msg.clone()),
+        }
+    }
+
+    pub fn from_fusion(result: &DetectionResult, m: &RuleMatch) -> Self {
+        Self {
+            timestamp: now_secs(),
+            flow_key: result.flow_key.clone(),
+            src_ip: result.flow_key_raw.src_ip.to_string(),
+            dst_ip: result.flow_key_raw.dst_ip.to_string(),
+            src_port: result.flow_key_raw.src_port,
+            dst_port: result.flow_key_raw.dst_port,
+            protocol: result.flow_key_raw.protocol,
+            source: AlertSource::Fusion,
+            severity: AlertSeverity::Critical,
+            is_attack: result.is_attack,
+            attack_type: result.attack_type.clone(),
+            confidence: result.confidence,
+            ae_score: result.ae_score,
+            rule_sid: Some(m.sid),
+            rule_msg: Some(m.msg.clone()),
+        }
+    }
+
+    pub fn from_tcp_anomaly(event: &TcpAnomalyEvent) -> Self {
+        let src_ip = format_ipv4(event.src_ip).to_string();
+        let dst_ip = format_ipv4(event.dst_ip).to_string();
+        let flow_key = format!("{}:{}-{}:{}", src_ip, event.src_port, dst_ip, event.dst_port);
+        let direction = if event.direction == DIRECTION_INGRESS { "ingress" } else { "egress" };
+        let attack_type = match event.anomaly_type {
+            ANOMALY_SYN_FIN => format!("TCP SYN+FIN anomaly ({})", direction),
+            ANOMALY_NULL_SCAN => format!("TCP NULL scan ({})", direction),
+            ANOMALY_XMAS_SCAN => format!("TCP XMAS scan ({})", direction),
+            ANOMALY_RST_SYN => format!("TCP RST+SYN anomaly ({})", direction),
+            _ => format!("TCP flag anomaly ({})", direction),
+        };
+        Self {
+            timestamp: now_secs(),
+            flow_key,
+            src_ip,
+            dst_ip,
+            src_port: event.src_port,
+            dst_port: event.dst_port,
+            protocol: 6,
+            source: AlertSource::TcpAnomaly,
+            severity: AlertSeverity::High,
+            is_attack: true,
+            attack_type: Some(attack_type),
+            confidence: 1.0,
+            ae_score: 0.0,
+            rule_sid: None,
+            rule_msg: None,
+        }
+    }
+}

mantis/src/model/ml_detection.rs

@@ -3,7 +3,6 @@ use compact_str::CompactString;
 use serde::{Deserialize, Serialize};
 
 use crate::model::direction::Direction;
-use crate::model::rule_detection::RuleMatch;
 use crate::utils::packet_parser::{format_ipv4, format_ipv6};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -11,6 +10,7 @@ pub struct ClipParams {
     pub lower: f64,
     pub upper: f64,
 }
+
 #[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
 pub struct FlowKey {
     pub src_ip: CompactString,
@@ -120,121 +120,3 @@ impl InferenceStats {
 pub struct EngineStats {
     pub active_flows: usize,
 }
-
-#[derive(Debug, Clone, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub enum AlertSource {
-    Ml,
-    Rule,
-    Fusion,
-}
-
-#[derive(Debug, Clone, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub enum AlertSeverity {
-    /// Single-source detection: either ML or Rule fired alone.
-    High,
-    /// Both ML and Rule agreed on the same flow within the fusion window.
-    Critical,
-}
-
-#[derive(Debug, Clone, Serialize)]
-pub struct UnifiedAlert {
-    pub timestamp: u64,
-    pub flow_key: String,
-    pub src_ip: String,
-    pub dst_ip: String,
-    pub src_port: u16,
-    pub dst_port: u16,
-    pub protocol: u8,
-    pub source: AlertSource,
-    pub severity: AlertSeverity,
-    pub is_attack: bool,
-    pub attack_type: Option<String>,
-    pub confidence: f32,
-    pub ae_score: f32,
-    pub rule_sid: Option<u32>,
-    pub rule_msg: Option<String>,
-}
-
-fn now_secs() -> u64 {
-    std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .expect("system time is after UNIX_EPOCH")
-        .as_secs()
-}
-
-fn parse_ip_port(s: &str) -> (String, u16) {
-    match s.rfind(':') {
-        Some(pos) => {
-            let port = s[pos + 1..].parse::<u16>().unwrap_or(0);
-            (s[..pos].to_string(), port)
-        }
-        None => (s.to_string(), 0),
-    }
-}
-
-impl UnifiedAlert {
-    pub fn from_ml(result: &DetectionResult) -> Self {
-        Self {
-            timestamp: now_secs(),
-            flow_key: result.flow_key.clone(),
-            src_ip: result.flow_key_raw.src_ip.to_string(),
-            dst_ip: result.flow_key_raw.dst_ip.to_string(),
-            src_port: result.flow_key_raw.src_port,
-            dst_port: result.flow_key_raw.dst_port,
-            protocol: result.flow_key_raw.protocol,
-            source: AlertSource::Ml,
-            severity: AlertSeverity::High,
-            is_attack: result.is_attack,
-            attack_type: result.attack_type.clone(),
-            confidence: result.confidence,
-            ae_score: result.ae_score,
-            rule_sid: None,
-            rule_msg: None,
-        }
-    }
-
-    pub fn from_rule(m: &RuleMatch) -> Self {
-        let (src_ip, src_port) = parse_ip_port(&m.src);
-        let (dst_ip, dst_port) = parse_ip_port(&m.dst);
-        let flow_key = format!("{}->{}", m.src, m.dst);
-        Self {
-            timestamp: now_secs(),
-            flow_key,
-            src_ip,
-            dst_ip,
-            src_port,
-            dst_port,
-            protocol: 6,
-            source: AlertSource::Rule,
-            severity: AlertSeverity::High,
-            is_attack: true,
-            attack_type: None,
-            confidence: 1.0,
-            ae_score: 0.0,
-            rule_sid: Some(m.sid),
-            rule_msg: Some(m.msg.clone()),
-        }
-    }
-
-    pub fn from_fusion(result: &DetectionResult, m: &RuleMatch) -> Self {
-        Self {
-            timestamp: now_secs(),
-            flow_key: result.flow_key.clone(),
-            src_ip: result.flow_key_raw.src_ip.to_string(),
-            dst_ip: result.flow_key_raw.dst_ip.to_string(),
-            src_port: result.flow_key_raw.src_port,
-            dst_port: result.flow_key_raw.dst_port,
-            protocol: result.flow_key_raw.protocol,
-            source: AlertSource::Fusion,
-            severity: AlertSeverity::Critical,
-            is_attack: result.is_attack,
-            attack_type: result.attack_type.clone(),
-            confidence: result.confidence,
-            ae_score: result.ae_score,
-            rule_sid: Some(m.sid),
-            rule_msg: Some(m.msg.clone()),
-        }
-    }
-}

mantis/src/model/mod.rs

@@ -1,3 +1,4 @@
+pub mod alert;
 pub mod config;
 pub mod direction;
 pub mod error;

mantis/src/web/websocket/alert_websocket.rs

@@ -6,7 +6,7 @@ use tokio::sync::broadcast;
 use crate::model::error::http::HttpError;
 use crate::model::error::misc::MiscError;
 use crate::model::log::http::HttpLog;
-use crate::model::ml_detection::UnifiedAlert;
+use crate::model::alert::UnifiedAlert;
 
 pub async fn handle_alert(socket: WebSocket, mut broadcast_rx: broadcast::Receiver<UnifiedAlert>) {
     let (mut sender, mut receiver) = socket.split();