Add TCP flag anomaly detection via XDP + BPF ring buffer

Detects four degenerate TCP flag combinations in XDP (ingress and egress):
- NULL scan (no flags set)
- SYN+FIN (impossible in valid TCP)
- XMAS scan (FIN+PSH+URG)
- RST+SYN

Anomalous packets are dropped at XDP and an event is written to a per-direction
BPF_MAP_TYPE_RINGBUF (TCP_ANOMALY_EVENTS, 256 KB each).  Userspace reads both
ring buffers via AsyncFd + tokio::select and broadcasts a UnifiedAlert with
source=TcpAnomaly through the existing DetectionAlert channel so WebSocket
clients receive real-time alerts.

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

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/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::ml_detection::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/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/model/ml_detection.rs

@@ -1,4 +1,8 @@
 use common::model::event::{Event, TcpFlags};
+use common::model::tcp_anomaly::{
+    ANOMALY_NULL_SCAN, ANOMALY_RST_SYN, ANOMALY_SYN_FIN, ANOMALY_XMAS_SCAN, DIRECTION_INGRESS,
+    TcpAnomalyEvent,
+};
 use compact_str::CompactString;
 use serde::{Deserialize, Serialize};
 
@@ -127,6 +131,7 @@ pub enum AlertSource {
     Ml,
     Rule,
     Fusion,
+    TcpAnomaly,
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -237,4 +242,35 @@ impl UnifiedAlert {
             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,
+        }
+    }
 }