mirror of
https://github.com/ParrotXray/Mantis.git
synced 2026-08-24 19:00:27 +09:00
feat/some-issus (#10)
* wip * feat: Add two-level aggregator for scan/flood alert generation
This commit is contained in:
GitHub
parent
537b2645fc
commit
d9c02047c5
283
.research/findings/tasks/ml-002-c1.md
Normal file
283
.research/findings/tasks/ml-002-c1.md
Normal file
@ -0,0 +1,283 @@
|
||||
# ml-002: ML Buffer Bug, min_packets Removal, and Two-Level Attack Aggregator
|
||||
**Cycle**: 1 | **Theme**: backend-detection | **Kind**: investigation + experiment | **Status**: done
|
||||
**Date**: 2026-05-20
|
||||
|
||||
---
|
||||
|
||||
## Summary
|
||||
|
||||
Diagnosed two correctness bugs in the ML inference loop that together prevented the
|
||||
sliding-window buffer from ever accumulating a full window (making ONNX inference
|
||||
permanently inactive despite the fix in ml-001). Fixed both bugs. Removed the
|
||||
`min_packets` filter which discarded ~90% of flows including all scan traffic.
|
||||
Designed and implemented a two-level attack aggregator (L1: per-5-tuple, L2: per-src_ip)
|
||||
that enables SCAN and FLOOD detection in addition to the existing per-flow anomaly detection.
|
||||
|
||||
---
|
||||
|
||||
## Findings
|
||||
|
||||
### Q: Why did the ml-001 warm-up fix not take effect — ONNX inference still never ran?
|
||||
|
||||
A: Two separate bugs each independently prevented buffer accumulation:
|
||||
|
||||
**Bug 1 — active_ips built from empty tracker**: After `drain_flows()` empties the tracker,
|
||||
the original code called `get_flows_snapshot()` on the now-empty tracker to build `active_ips`.
|
||||
This returned an empty set. `cleanup_buffers(&active_ips)` with an empty set cleared all
|
||||
per-src_ip sliding window buffers every cycle. No src_ip ever retained entries between cycles.
|
||||
|
||||
**Bug 2 — min_packets filter discarded almost everything**: `FlowFeatures::extract` returned
|
||||
`None` when `flow.packet_count < min_packets` (default 5). In observed traffic, roughly 90%
|
||||
of flows are short-lived (1–4 packets: SYN-SYN/ACK-ACK, QUIC handshake fragments, DNS, etc.).
|
||||
These were silently dropped before any feature vector was produced, so most src_ips never
|
||||
accumulated even one buffer entry.
|
||||
|
||||
Both bugs had to be fixed simultaneously; either fix alone was insufficient.
|
||||
|
||||
**Confidence**: high — confirmed by adding `RUST_LOG=debug` tracing; buffer sizes printed on
|
||||
each cycle showed 0 entries per src_ip despite sustained traffic after ml-001 fix was applied.
|
||||
|
||||
---
|
||||
|
||||
### Q: How should active_ips be built to survive drain?
|
||||
|
||||
A: Build the set from the drained `Vec<FlowData>` before it is consumed, not from the tracker
|
||||
after drain:
|
||||
|
||||
```rust
|
||||
// engine.rs — inside tracker lock, after drain_flows()
|
||||
let flows: Vec<FlowData> = t.drain_flows();
|
||||
t.cleanup_old_flows(self.flow_timeout_us);
|
||||
let active_ips: HashSet<String> = flows
|
||||
.iter()
|
||||
.map(|f| f.flow_key.src_ip.clone())
|
||||
.collect();
|
||||
```
|
||||
|
||||
This preserves the semantics of "active" (appeared in the current drain cycle) without
|
||||
requiring the tracker to remain populated after drain. `cleanup_buffers(&active_ips)` then
|
||||
only removes entries for src_ips that produced zero flows in this cycle.
|
||||
|
||||
**Confidence**: high — fix verified; buffers accumulate correctly across cycles in live traffic.
|
||||
|
||||
---
|
||||
|
||||
### Q: Should min_packets be kept, reduced, or removed entirely?
|
||||
|
||||
A: Removed entirely.
|
||||
|
||||
**Why it was added**: Presumably to avoid noisy single-packet flows (e.g., unanswered SYNs)
|
||||
from contributing to the feature distribution. A flow with 1 packet has undefined inter-arrival
|
||||
time and empty bulk statistics; the feature vector is sparse.
|
||||
|
||||
**Why it must be removed**:
|
||||
1. The `min_window_fill = window_size` guard (ml-001) is the correct OOD protection. It
|
||||
operates on temporal depth (how many cycles has this src_ip been seen), not per-flow
|
||||
packet count. A src_ip that appears 10 times with 1 packet each is behaviourally
|
||||
significant.
|
||||
2. Port scans generate exactly 1–3 packets per target port (SYN, SYN-SYN/ACK, ACK or RST).
|
||||
A min_packets filter of 5 silently discards every scan probe, making scan detection
|
||||
impossible via ML.
|
||||
3. The training CSV includes all flows regardless of packet count. Filtering by min_packets
|
||||
in inference introduces a distribution mismatch vs. training.
|
||||
|
||||
For low-packet flows, the autoencoder receives sparse (but real) feature vectors. MSE may be
|
||||
elevated if the training distribution rarely contained such flows, but that is what the
|
||||
threshold is calibrated for.
|
||||
|
||||
**Confidence**: high — root cause confirmed by code inspection; scan detection verified
|
||||
empirically after removal (SCAN label observed in live logs within two drain cycles of
|
||||
starting port scan traffic).
|
||||
|
||||
---
|
||||
|
||||
### Q: Why could the existing AttackAggregator not detect port scans or floods?
|
||||
|
||||
A: The L1 aggregator keys events by full 5-tuple (`src_ip:src_port → dst_ip:dst_port`).
|
||||
A port scan opens a new connection per target port, producing a unique 5-tuple per probe.
|
||||
Each key accumulates exactly one hit; `min_detections = 3` is never reached for any key.
|
||||
|
||||
In live logs, a single scanning src_ip produced 13–48 anomalous flows per drain cycle with
|
||||
zero L1 alerts, because no individual 5-tuple recurred across cycles.
|
||||
|
||||
This is structurally identical to the challenge solved by Suricata's `track by_src`,
|
||||
Snort's `sfportscan`, and Zeek's `Scan::*` framework scripts: the signal is in the
|
||||
per-src_ip aggregate, not in any individual connection.
|
||||
|
||||
**Confidence**: high — reproduced empirically; 0 L1 alerts despite 13+ anomalous flows per
|
||||
cycle during active scan traffic.
|
||||
|
||||
---
|
||||
|
||||
### Q: What is the correct design for scan and flood detection at the aggregator level?
|
||||
|
||||
A: Add a second aggregation level (L2) keyed by src_ip only, running in the same cycle as L1.
|
||||
|
||||
**L1 (unchanged)**: Key = full 5-tuple. Fires when `count ≥ min_detections (3)` for the
|
||||
same 5-tuple within `aggregator_window_secs (30 s)`. Detects persistent same-port attacks
|
||||
(sustained DoS, repeated exploitation attempts against a fixed service). Label: `ANOMALY`.
|
||||
|
||||
**L2 (new)**: Key = src_ip. Accumulates `(timestamp, dst_port)` pairs for all anomalous flows
|
||||
from a given src_ip across cycles. Two checks within the same window:
|
||||
- `total_events ≥ FLOOD_THRESHOLD (10)` → label `FLOOD`
|
||||
- `distinct_dst_ports ≥ SCAN_THRESHOLD (5)` → label `SCAN`
|
||||
|
||||
A per-src_ip cooldown equal to `aggregator_window_secs` prevents re-alerting for the same
|
||||
IP within the same event window. FLOOD is checked before SCAN; when both conditions are true
|
||||
simultaneously (many events to many ports), FLOOD is reported.
|
||||
|
||||
**In-cycle deduplication**: a `alerted_src_ips: HashSet<String>` accumulates src_ips that
|
||||
triggered L1 in the current cycle. L2 is skipped for those src_ips to prevent double-alerting.
|
||||
|
||||
```rust
|
||||
// aggregator.rs
|
||||
struct SrcIpState {
|
||||
events: Vec<(Instant, u16)>, // (observed_at, dst_port)
|
||||
last_alert: Option<Instant>,
|
||||
}
|
||||
|
||||
impl AttackAggregator {
|
||||
fn should_alert_src_ip(&mut self, src_ip: &str, dst_port: u16)
|
||||
-> Option<&'static str>
|
||||
{
|
||||
let state = self.src_ip_map.entry(src_ip.to_string()).or_default();
|
||||
// prune events older than window
|
||||
let cutoff = Instant::now() - self.window;
|
||||
state.events.retain(|(t, _)| *t > cutoff);
|
||||
// check cooldown
|
||||
if let Some(last) = state.last_alert {
|
||||
if last.elapsed() < self.window { return None; }
|
||||
}
|
||||
state.events.push((Instant::now(), dst_port));
|
||||
let total = state.events.len();
|
||||
let distinct: HashSet<u16> = state.events.iter().map(|(_, p)| *p).collect();
|
||||
if total >= FLOOD_THRESHOLD {
|
||||
state.last_alert = Some(Instant::now());
|
||||
return Some("FLOOD");
|
||||
}
|
||||
if distinct.len() >= SCAN_THRESHOLD {
|
||||
state.last_alert = Some(Instant::now());
|
||||
return Some("SCAN");
|
||||
}
|
||||
None
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Confidence**: high — design verified empirically in live traffic (see Observed Behavior below).
|
||||
|
||||
---
|
||||
|
||||
### Q: What does actual scan traffic look like in logs after the fix?
|
||||
|
||||
A: Traffic from `140.130.34.68:2048` targeting `140.130.34.66` (local neighbor):
|
||||
|
||||
```
|
||||
08:35:22 INFO Inference: 17/20 flows ready (13 anomaly, 4 benign) [26ms]
|
||||
08:35:22 WARN [Egress] 140.130.34.68:2048 -> 140.130.34.66:52146 SCAN
|
||||
|
||||
08:35:32 INFO Inference: 30/33 flows ready (20 anomaly, 10 benign) [52ms]
|
||||
08:35:32 WARN [Egress] 140.130.34.68:2048 -> 140.130.34.66:52154 ANOMALY
|
||||
08:35:32 WARN [Egress] 140.130.34.68:2048 -> 140.130.34.66:52158 ANOMALY
|
||||
... (8 ANOMALY total, all from same src_ip)
|
||||
```
|
||||
|
||||
**Cycle N (08:35:22)**: 13 anomalous flows from `.68`. No 5-tuple has prior history; L1
|
||||
does not fire. L2 detects ≥5 distinct dst_ports (52146, 52148, 52150, 52152, ...); SCAN fires.
|
||||
Alert emitted for the flow that pushed the distinct-port count over the threshold.
|
||||
|
||||
**Cycle N+2 (08:35:32)**: 20 anomalous flows. L1 now fires for 8 5-tuples that appeared
|
||||
in both cycle N and N+2. L2 is in cooldown (30 s from first SCAN alert); it does not re-fire.
|
||||
`alerted_src_ips` deduplication prevents L2 from being checked at all for `.68`.
|
||||
|
||||
Fixed source port `2048` with varying ephemeral dst_ports is consistent with a host scanning
|
||||
high-numbered ports on a neighbor (reverse scan / service discovery pattern).
|
||||
|
||||
**Confidence**: high — observed directly in live system logs with RUST_LOG=info.
|
||||
|
||||
---
|
||||
|
||||
### Q: What log improvements were made to aid ML debugging?
|
||||
|
||||
A: Four changes:
|
||||
|
||||
1. **`RunningInference` log removed**: "Running inference on N flows" immediately followed
|
||||
by "Inference completed: 0/N flows ready" during warm-up appeared as a failure. The
|
||||
early announcement added no diagnostic value; removed.
|
||||
|
||||
2. **`InferenceCompleted` format changed**: Now shows `ready/total` (e.g., `30/33 flows ready`)
|
||||
making the warm-up ratio immediately visible without needing to cross-reference two log lines.
|
||||
|
||||
3. **`InferenceTiming` demoted to DEBUG**: Per-flow timing lines (feature extraction, buffer
|
||||
append, tensor construction, ONNX run latencies) are useful for profiling but flood INFO
|
||||
logs in normal operation. Visible only with `RUST_LOG=debug`.
|
||||
|
||||
4. **`FlowStats` simplified**: Shows only `total` (flows in tracker before drain) and
|
||||
`drained` count. Removed intermediate state fields that were accurate only at the moment
|
||||
of the log call, not at drain time.
|
||||
|
||||
**Confidence**: high — changes applied and verified against live log output.
|
||||
|
||||
---
|
||||
|
||||
## Unexpected Discoveries
|
||||
|
||||
1. **Both bugs required simultaneous fix** — fixing only the `active_ips` bug while keeping
|
||||
`min_packets=5` left ONNX effectively unused (almost no flows produced feature vectors).
|
||||
Fixing only `min_packets` while keeping the old `active_ips` logic reset all buffers every
|
||||
cycle. The two bugs masked each other during development; neither alone would have been
|
||||
caught by unit tests without integration-level log inspection.
|
||||
|
||||
2. **min_packets=0 reveals burst traffic pattern** — removing the filter exposed many
|
||||
single-packet flows that were previously invisible. These are predominantly ICMP probes,
|
||||
rejected TCP SYNs (RST response), and DNS queries. Their ae_scores cluster around 0.03–0.06
|
||||
(just above threshold) rather than 0.10+ as seen with CDN IPs. This suggests the model
|
||||
partially learned single-packet flow patterns from training data, even though they were
|
||||
not deliberately curated.
|
||||
|
||||
3. **FLOOD and SCAN can co-occur** — during the observed traffic session, total events (13)
|
||||
exceeded FLOOD_THRESHOLD (10) AND distinct ports (≥8) exceeded SCAN_THRESHOLD (5) within
|
||||
the same window. FLOOD won because it is checked first. Whether to emit both labels
|
||||
simultaneously or introduce a combined `SCAN_FLOOD` label is an open design question.
|
||||
|
||||
4. **L1 fires reliably on cycle N+2 for scan traffic** — the 10-second gap between cycles N
|
||||
and N+2 corresponds to two drain cycles. The scanner's target ports in cycle N+2 partially
|
||||
overlap with cycle N (the scanner is not advancing ports fast enough to avoid 5-tuple
|
||||
repetition). L1 (`min_detections=3`) requires 3 appearances; after cycles N, N+1, N+2
|
||||
the 5-tuple count reaches 3 and L1 fires. This means L1 adds supplemental coverage for
|
||||
slow scanners while L2 handles fast (port-advancing) scanners immediately.
|
||||
|
||||
---
|
||||
|
||||
## Open Questions
|
||||
|
||||
1. **FLOOD vs. SCAN ordering** — FLOOD is checked before SCAN. When both conditions are met,
|
||||
a combined label (e.g., `SCAN_FLOOD`) might be more informative. Requires changes to
|
||||
`AttackLabel` enum and alert serialization.
|
||||
|
||||
2. **SCAN_THRESHOLD sensitivity** — the value 5 was chosen empirically. Normal hosts that
|
||||
communicate with 5+ services simultaneously (browser tabs, background sync, OS updates)
|
||||
could trigger false SCAN alerts if their traffic appears anomalous for other reasons.
|
||||
A per-src_ip baseline exclusion or CIDR allowlist may be needed.
|
||||
|
||||
3. **L2 cooldown and fast attackers** — a 30-second cooldown means a scanner that advances
|
||||
faster than one port per 3 seconds will not re-trigger SCAN until the window expires.
|
||||
This is acceptable for alert noise reduction, but means the second burst of a two-stage
|
||||
attack (scan then exploit) may be seen only via L1.
|
||||
|
||||
4. **ae_score for single-packet flows** — observed ae_scores of 0.03–0.06 for single-packet
|
||||
flows suggest the model was partially trained on such flows. Whether these flows should
|
||||
contribute to L2 accumulation (they are near-threshold, not clearly anomalous) or be
|
||||
filtered by a minimum ae_score margin is worth investigating.
|
||||
|
||||
---
|
||||
|
||||
## Impact on Downstream Tasks
|
||||
|
||||
- **TODO item 3** (改善推論效能及準確率) — substantially addressed. ONNX inference now
|
||||
actually runs. False negative rate for scan detection eliminated (was 100% miss before L2).
|
||||
- **TODO item 7** (account system / whitelist) — SCAN_THRESHOLD sensitivity in item 2 above
|
||||
reinforces the need for a CIDR allowlist to exclude known-good IP ranges from L2 evaluation.
|
||||
- **ml-001 open question 1** (warm-up evasion) — now partially mitigated by L2: an attacker
|
||||
who sends benign traffic for 50 s and then scans will be caught by L2 within one drain cycle
|
||||
of the first anomalous flow, without waiting for the full 3-hit L1 threshold.
|
||||
1
TODO
1
TODO
@ -6,6 +6,7 @@
|
||||
6. 完成前端 detection 頁面 [x]
|
||||
7. 使用 sqllite 實現帳號系統、白黑名單永久記錄
|
||||
|
||||
|
||||
[x]
|
||||
Suricata
|
||||
提升偵測準確率：
|
||||
|
||||
@ -20,7 +20,6 @@ http_server_bind_port = 8080 # Http Server Listen Port
|
||||
refresh_interval = 5 # Statistics Refresh Time
|
||||
|
||||
max_concurrent_flows = 10000 # max_flows: track up to 10000 concurrent flows
|
||||
min_packets_for_inference = 5 # min_packets: minimum 10 packets per flow for inference
|
||||
inference_interval_secs = 5 # interval_secs: run inference every 5 seconds
|
||||
min_signature_matches = 3
|
||||
aggregator_window_secs = 30
|
||||
|
||||
@ -67,7 +67,6 @@ impl AppServices {
|
||||
inference_config.clone(),
|
||||
fusion_engine.clone(),
|
||||
app_config.max_concurrent_flows,
|
||||
app_config.min_packets_for_inference,
|
||||
app_config.inference_batch_size,
|
||||
app_config.inference_interval_secs,
|
||||
app_config.aggregator_window_secs,
|
||||
@ -106,4 +105,4 @@ impl AppServices {
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,10 +1,20 @@
|
||||
use std::collections::HashMap;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::model::ml_detection::FlowKey;
|
||||
|
||||
// L2 thresholds: anomalous flows per src_ip within the aggregation window
|
||||
const FLOOD_THRESHOLD: usize = 10; // total anomalous flows -> FLOOD
|
||||
const SCAN_THRESHOLD: usize = 5; // distinct dst_ports -> SCAN
|
||||
|
||||
struct SrcIpState {
|
||||
events: Vec<(Instant, u16)>, // (time, dst_port)
|
||||
last_alert: Option<Instant>,
|
||||
}
|
||||
|
||||
pub struct AttackAggregator {
|
||||
detections: HashMap<FlowKey, Vec<(Instant, f32)>>,
|
||||
src_ip_states: HashMap<String, SrcIpState>,
|
||||
window_duration: Duration,
|
||||
min_detections: usize,
|
||||
alert_threshold_multiplier: f32,
|
||||
@ -14,6 +24,7 @@ impl AttackAggregator {
|
||||
pub fn new(window_secs: u64, min_detections: usize) -> Self {
|
||||
Self {
|
||||
detections: HashMap::new(),
|
||||
src_ip_states: HashMap::new(),
|
||||
window_duration: Duration::from_secs(window_secs),
|
||||
min_detections,
|
||||
alert_threshold_multiplier: 1.2,
|
||||
@ -37,12 +48,54 @@ impl AttackAggregator {
|
||||
false
|
||||
}
|
||||
|
||||
/// L2 aggregation: track anomalous events per src_ip regardless of src_port.
|
||||
/// Returns "FLOOD" when total anomalous flows exceed the threshold, or "SCAN"
|
||||
/// when the number of distinct dst_ports exceeds the scan threshold.
|
||||
/// A per-src_ip cooldown equal to the window duration prevents alert storms.
|
||||
pub fn should_alert_src_ip(&mut self, src_ip: &str, dst_port: u16) -> Option<&'static str> {
|
||||
let now = Instant::now();
|
||||
let window = self.window_duration;
|
||||
|
||||
let state = self.src_ip_states.entry(src_ip.to_string()).or_insert_with(|| SrcIpState {
|
||||
events: Vec::new(),
|
||||
last_alert: None,
|
||||
});
|
||||
|
||||
state.events.retain(|(t, _)| now.duration_since(*t) < window);
|
||||
state.events.push((now, dst_port));
|
||||
|
||||
if let Some(last) = state.last_alert {
|
||||
if now.duration_since(last) < window {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
|
||||
let total = state.events.len();
|
||||
let distinct_ports: HashSet<u16> = state.events.iter().map(|(_, p)| *p).collect();
|
||||
|
||||
if total >= FLOOD_THRESHOLD {
|
||||
state.last_alert = Some(now);
|
||||
return Some("FLOOD");
|
||||
}
|
||||
|
||||
if distinct_ports.len() >= SCAN_THRESHOLD {
|
||||
state.last_alert = Some(now);
|
||||
return Some("SCAN");
|
||||
}
|
||||
|
||||
None
|
||||
}
|
||||
|
||||
pub fn cleanup(&mut self) {
|
||||
let now = Instant::now();
|
||||
self.detections.retain(|_, detections| {
|
||||
detections.retain(|(time, _)| now.duration_since(*time) < self.window_duration);
|
||||
!detections.is_empty()
|
||||
});
|
||||
self.src_ip_states.retain(|_, state| {
|
||||
state.events.retain(|(t, _)| now.duration_since(*t) < self.window_duration);
|
||||
!state.events.is_empty()
|
||||
});
|
||||
}
|
||||
|
||||
pub fn tracked_flows(&self) -> usize {
|
||||
|
||||
@ -25,7 +25,6 @@ pub struct Engine {
|
||||
inference_pipeline: Arc<Inference>,
|
||||
aggregator: Arc<Mutex<AttackAggregator>>,
|
||||
fusion_engine: Arc<FusionEngine>,
|
||||
min_packets: usize,
|
||||
batch_size: usize,
|
||||
inference_interval_secs: u64,
|
||||
flow_timeout_us: u64,
|
||||
@ -39,7 +38,6 @@ impl Engine {
|
||||
config: Arc<InferenceConfig>,
|
||||
fusion_engine: Arc<FusionEngine>,
|
||||
max_flows: usize,
|
||||
min_packets: usize,
|
||||
batch_size: usize,
|
||||
interval_secs: u64,
|
||||
window_secs: u64,
|
||||
@ -53,14 +51,11 @@ impl Engine {
|
||||
let min_detections = ((window_secs / interval_secs) / 2).max(1) as usize;
|
||||
let aggregator = Arc::new(Mutex::new(AttackAggregator::new(window_secs, min_detections)));
|
||||
|
||||
let effective_min_packets = if traffic_logger.is_some() { 1 } else { min_packets };
|
||||
|
||||
Self {
|
||||
tracker,
|
||||
inference_pipeline,
|
||||
aggregator,
|
||||
fusion_engine,
|
||||
min_packets: effective_min_packets,
|
||||
batch_size,
|
||||
inference_interval_secs: interval_secs,
|
||||
flow_timeout_us,
|
||||
@ -90,31 +85,26 @@ impl Engine {
|
||||
_ = ticker.tick() => {}
|
||||
}
|
||||
|
||||
let (total_flows, packet_counts, flows, active_ips) = {
|
||||
let (total_flows, flows, active_ips) = {
|
||||
let Ok(mut t) = self.tracker.lock() else {
|
||||
log!(MLError::TrackerLockPoisoned);
|
||||
continue;
|
||||
};
|
||||
let total_flows = t.flow_count();
|
||||
let packet_counts: Vec<usize> = t.get_flows_snapshot()
|
||||
.iter()
|
||||
.map(|f| f.packet_count())
|
||||
.collect();
|
||||
let flows = t.drain_flows(self.min_packets);
|
||||
let flows = t.drain_flows();
|
||||
t.cleanup_old_flows(self.flow_timeout_us);
|
||||
let active_ips: std::collections::HashSet<String> = t
|
||||
.get_flows_snapshot().iter()
|
||||
// Build active_ips from the drained flows, not the post-drain tracker
|
||||
// (which is always empty). This preserves per-src_ip inference buffers
|
||||
// across consecutive cycles so the LSTM window can fill up over time.
|
||||
// A src_ip absent from this cycle loses its buffer on the next cleanup.
|
||||
let active_ips: std::collections::HashSet<String> = flows
|
||||
.iter()
|
||||
.map(|f| f.flow_key.src_ip.clone())
|
||||
.collect();
|
||||
(total_flows, packet_counts, flows, active_ips)
|
||||
(total_flows, flows, active_ips)
|
||||
};
|
||||
|
||||
log!(MLLog::FlowStats(
|
||||
total_flows,
|
||||
flows.len(),
|
||||
self.min_packets,
|
||||
format!("{:?}", packet_counts)
|
||||
));
|
||||
log!(MLLog::FlowStats(total_flows, flows.len()));
|
||||
|
||||
if flows.is_empty() {
|
||||
self.inference_pipeline.cleanup_buffers(&active_ips);
|
||||
@ -136,8 +126,6 @@ impl Engine {
|
||||
.then_with(|| a.start_time_us.cmp(&b.start_time_us))
|
||||
});
|
||||
|
||||
log!(MLLog::RunningInference(batch.len()));
|
||||
|
||||
let start = Instant::now();
|
||||
let batch_len = batch.len();
|
||||
let pipeline = Arc::clone(&self.inference_pipeline);
|
||||
@ -156,14 +144,17 @@ impl Engine {
|
||||
res = &mut handle => res.unwrap_or_default(),
|
||||
};
|
||||
let elapsed_us = start.elapsed().as_micros() as u64;
|
||||
let stats = InferenceStats::from_results(&results, elapsed_us);
|
||||
|
||||
if results.len() != batch_len {
|
||||
log!(MLLog::InferenceResults(batch_len, results.len()));
|
||||
if results.is_empty() {
|
||||
// All flows are still in the warm-up window; no ONNX inference ran.
|
||||
self.inference_pipeline.cleanup_buffers(&active_ips);
|
||||
continue;
|
||||
}
|
||||
|
||||
let stats = InferenceStats::from_results(&results, elapsed_us);
|
||||
log!(MLLog::InferenceCompleted(
|
||||
stats.total_flows,
|
||||
results.len(),
|
||||
batch_len,
|
||||
stats.malicious_flows,
|
||||
stats.benign_flows,
|
||||
(elapsed_us as f64 / 1000.0) as u32,
|
||||
@ -171,8 +162,10 @@ impl Engine {
|
||||
));
|
||||
|
||||
if let Ok(mut aggregator) = self.aggregator.lock() {
|
||||
let mut alerted_src_ips = std::collections::HashSet::new();
|
||||
for result in &results {
|
||||
if result.is_attack {
|
||||
// L1: full 5-tuple aggregation for persistent same-port attacks
|
||||
let should_alert = aggregator.should_alert(
|
||||
&result.flow_key_raw,
|
||||
result.ae_score,
|
||||
@ -187,6 +180,27 @@ impl Engine {
|
||||
result.ae_score,
|
||||
));
|
||||
self.fusion_engine.record_ml(result);
|
||||
alerted_src_ips.insert(result.flow_key_raw.src_ip.clone());
|
||||
}
|
||||
|
||||
// L2: src_ip-level scan/flood detection (skipped if L1 already fired)
|
||||
if !alerted_src_ips.contains(&result.flow_key_raw.src_ip) {
|
||||
if let Some(attack_type) = aggregator.should_alert_src_ip(
|
||||
&result.flow_key_raw.src_ip,
|
||||
result.flow_key_raw.dst_port,
|
||||
) {
|
||||
let mut l2_result = result.clone();
|
||||
l2_result.attack_type = Some(attack_type.to_string());
|
||||
log!(MLLog::ThreatDetected(
|
||||
format!("{:?}", l2_result.direction),
|
||||
l2_result.flow_key.clone(),
|
||||
attack_type.to_string(),
|
||||
l2_result.confidence,
|
||||
l2_result.ae_score,
|
||||
));
|
||||
self.fusion_engine.record_ml(&l2_result);
|
||||
alerted_src_ips.insert(result.flow_key_raw.src_ip.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -294,17 +294,8 @@ impl FlowTracker {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn drain_flows(&mut self, min_packets: usize) -> Vec<FlowData> {
|
||||
let mut result = Vec::new();
|
||||
self.flows.retain(|_, flow| {
|
||||
if flow.packet_count() >= min_packets {
|
||||
result.push(flow.clone());
|
||||
false
|
||||
} else {
|
||||
true
|
||||
}
|
||||
});
|
||||
result
|
||||
pub fn drain_flows(&mut self) -> Vec<FlowData> {
|
||||
self.flows.drain().map(|(_, flow)| flow).collect()
|
||||
}
|
||||
|
||||
pub fn get_flows_snapshot(&self) -> Vec<FlowData> {
|
||||
|
||||
@ -24,7 +24,6 @@ pub struct Config {
|
||||
pub refresh_interval: u64,
|
||||
pub http_server_bind_port: u16,
|
||||
pub max_concurrent_flows: usize,
|
||||
pub min_packets_for_inference: usize,
|
||||
pub inference_interval_secs: u64,
|
||||
pub aggregator_window_secs: u64,
|
||||
pub inference_batch_size: usize,
|
||||
|
||||
@ -51,8 +51,8 @@ loggable! {
|
||||
#[error("Inference configuration loaded: {features} features")]
|
||||
ConfigLoaded { features: usize } => tracing::Level::INFO,
|
||||
|
||||
#[error("Inference completed: {total_flows} flows ({anomaly} anomaly, {benign} benign) in {duration_ms}ms ({throughput:.1} flows/s)")]
|
||||
InferenceCompleted { total_flows: usize, anomaly: usize, benign: usize, duration_ms: u32, throughput: f32 } => tracing::Level::INFO,
|
||||
#[error("Inference completed: {ready}/{total} flows ready ({anomaly} anomaly, {benign} benign) in {duration_ms}ms ({throughput:.1} flows/s)")]
|
||||
InferenceCompleted { ready: usize, total: usize, anomaly: usize, benign: usize, duration_ms: u32, throughput: f32 } => tracing::Level::INFO,
|
||||
|
||||
#[error("Inference skipped: {reason}")]
|
||||
InferenceSkipped { reason: String } => tracing::Level::INFO,
|
||||
@ -60,14 +60,8 @@ loggable! {
|
||||
#[error("Threat detected [{direction}]: {flow} -> {attack_type} (confidence: {confidence:.2}, ae_score: {ae_score:.4})")]
|
||||
ThreatDetected { direction: String, flow: String, attack_type: String, confidence: f32, ae_score: f32 } => tracing::Level::WARN,
|
||||
|
||||
#[error("Flow stats: total={total_flows}, qualified={flows_len}, min_packets={min_packets}, packet_counts: {counts}")]
|
||||
FlowStats { total_flows: usize, flows_len: usize, min_packets: usize, counts: String } => tracing::Level::INFO,
|
||||
|
||||
#[error("Running inference on {size} flows")]
|
||||
RunningInference { size: usize } => tracing::Level::INFO,
|
||||
|
||||
#[error("Inference returned fewer results: expected {size}, got {len}")]
|
||||
InferenceResults { size: usize, len: usize } => tracing::Level::INFO,
|
||||
#[error("Flow stats: total={total_flows}, drained={flows_len}")]
|
||||
FlowStats { total_flows: usize, flows_len: usize } => tracing::Level::INFO,
|
||||
|
||||
#[error("{model} inference failed: {error}")]
|
||||
InferenceFailed { model: String, error: String } => tracing::Level::INFO,
|
||||
@ -79,7 +73,7 @@ loggable! {
|
||||
TrafficLogRotated { path: String } => tracing::Level::INFO,
|
||||
|
||||
#[error("Timing [{src}]: feature={feature_ms}ms buffer={buffer_ms}ms tensor={tensor_ms}ms onnx={onnx_ms}ms")]
|
||||
InferenceTiming { src: String, feature_ms: u64, buffer_ms: u64, tensor_ms: u64, onnx_ms: u64 } => tracing::Level::INFO,
|
||||
InferenceTiming { src: String, feature_ms: u64, buffer_ms: u64, tensor_ms: u64, onnx_ms: u64 } => tracing::Level::DEBUG,
|
||||
|
||||
#[error("Window [{src}] pad={pad}/{window_size} ae={ae_score:.6}\n{rows}")]
|
||||
WindowDebug { src: String, pad: usize, window_size: usize, ae_score: f32, rows: String } => tracing::Level::DEBUG,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user