#![cfg(all(feature = "emit-eve", feature = "tracker", feature = "extractors"))]
use flowscope::emit::{EveJsonWriter, EveOptions};
use flowscope::event::{AnomalyKind, EndReason, FlowEvent, FlowSide, FlowStats, OverflowPolicy};
use flowscope::extract::FiveTupleKey;
use flowscope::{HistoryString, L4Proto, Timestamp};
fn key() -> FiveTupleKey {
FiveTupleKey {
proto: L4Proto::Tcp,
a: "10.0.0.1:33000".parse().unwrap(),
b: "10.0.0.2:80".parse().unwrap(),
}
}
fn parse_lines(out: &[u8]) -> Vec<serde_json::Value> {
String::from_utf8(out.to_vec())
.unwrap()
.lines()
.map(|l| serde_json::from_str::<serde_json::Value>(l).unwrap())
.collect()
}
fn flipped_key() -> FiveTupleKey {
FiveTupleKey {
proto: L4Proto::Tcp,
a: "10.0.0.2:80".parse().unwrap(),
b: "10.0.0.1:33000".parse().unwrap(),
}
}
#[test]
fn eve_flow_hash_is_emitted_and_hex_format() {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key: key(),
kind: AnomalyKind::OutOfOrderSegment {
side: FlowSide::Initiator,
count: 1,
},
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
let lines = parse_lines(&buf);
let hash = lines[0]["flow_hash"].as_str().expect("flow_hash present");
assert_eq!(hash.len(), 16, "16-char hex u64");
assert!(
hash.chars().all(|c| c.is_ascii_hexdigit()),
"ascii hex: {hash}"
);
}
#[test]
fn eve_flow_hash_is_direction_invariant_and_deterministic() {
fn extract_hash(key: FiveTupleKey) -> String {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key,
kind: AnomalyKind::OutOfOrderSegment {
side: FlowSide::Initiator,
count: 1,
},
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
parse_lines(&buf)[0]["flow_hash"]
.as_str()
.unwrap()
.to_string()
}
let h1 = extract_hash(key());
let h2 = extract_hash(key()); let h3 = extract_hash(flipped_key()); assert_eq!(h1, h2);
assert_eq!(h1, h3);
}
#[test]
fn eve_flow_hash_differs_for_distinct_flows() {
fn extract_hash(key: FiveTupleKey) -> String {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key,
kind: AnomalyKind::OutOfOrderSegment {
side: FlowSide::Initiator,
count: 1,
},
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
parse_lines(&buf)[0]["flow_hash"]
.as_str()
.unwrap()
.to_string()
}
let h1 = extract_hash(key());
let mut other = key();
other.b = "10.0.0.3:443".parse().unwrap();
let h2 = extract_hash(other);
assert_ne!(h1, h2);
}
#[test]
fn eve_flow_anomaly_buffer_overflow_has_expected_shape() {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key: key(),
kind: AnomalyKind::BufferOverflow {
side: FlowSide::Initiator,
bytes: 1500,
policy: OverflowPolicy::SlidingWindow,
},
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
w.flush().unwrap();
let lines = parse_lines(&buf);
assert_eq!(lines.len(), 1);
let line = &lines[0];
assert_eq!(line["event_type"], "anomaly");
assert_eq!(line["src_ip"], "10.0.0.1");
assert_eq!(line["dest_ip"], "10.0.0.2");
assert_eq!(line["src_port"], 33000);
assert_eq!(line["dest_port"], 80);
assert_eq!(line["proto"], "TCP");
assert_eq!(line["anomaly"]["type"], "stream");
assert_eq!(line["anomaly"]["event"], "buffer_overflow");
assert_eq!(line["severity"], 3); assert!(line["timestamp"].as_str().unwrap().starts_with("2023-"));
}
#[test]
fn eve_session_parse_error_classifies_as_applayer() {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key: key(),
kind: AnomalyKind::SessionParseError {
side: FlowSide::Responder,
reason: Some("bad frame".to_string()),
},
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
let lines = parse_lines(&buf);
assert_eq!(lines[0]["anomaly"]["type"], "applayer");
assert_eq!(lines[0]["anomaly"]["event"], "parse_error");
assert_eq!(lines[0]["severity"], 2); }
#[test]
fn eve_flow_ended_includes_pkts_bytes_and_age() {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let mut stats = FlowStats::default();
stats.packets_initiator = 7;
stats.packets_responder = 5;
stats.bytes_initiator = 2400;
stats.bytes_responder = 14000;
stats.started = Timestamp::new(1_700_000_000, 0);
stats.last_seen = Timestamp::new(1_700_000_006, 0);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::Ended {
key: key(),
reason: EndReason::Fin,
stats,
history: HistoryString::new(),
l4: Some(L4Proto::Tcp),
};
w.write_event(&ev).unwrap();
let lines = parse_lines(&buf);
assert_eq!(lines[0]["event_type"], "flow");
assert_eq!(lines[0]["flow"]["pkts_toserver"], 7);
assert_eq!(lines[0]["flow"]["pkts_toclient"], 5);
assert_eq!(lines[0]["flow"]["bytes_toserver"], 2400);
assert_eq!(lines[0]["flow"]["bytes_toclient"], 14000);
assert_eq!(lines[0]["flow"]["age"], 6);
assert_eq!(lines[0]["flow"]["reason"], "fin");
assert_eq!(lines[0]["flow"]["alerted"], false);
}
#[test]
fn eve_flow_id_increases_monotonically() {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let kind = AnomalyKind::OutOfOrderSegment {
side: FlowSide::Initiator,
count: 1,
};
for _ in 0..3 {
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key: key(),
kind: kind.clone(),
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
}
let lines = parse_lines(&buf);
assert_eq!(lines[0]["flow_id"], 1);
assert_eq!(lines[1]["flow_id"], 2);
assert_eq!(lines[2]["flow_id"], 3);
}
#[test]
fn eve_anomalies_disabled_produces_no_output() {
let mut buf = Vec::new();
let mut opts = EveOptions::default();
opts.include_anomalies = false;
let mut w = EveJsonWriter::with_options(&mut buf, opts);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key: key(),
kind: AnomalyKind::OutOfOrderSegment {
side: FlowSide::Initiator,
count: 1,
},
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
assert!(buf.is_empty());
}
#[test]
fn eve_tick_off_by_default() {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::Tick {
key: key(),
stats: FlowStats::default(),
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
assert!(buf.is_empty(), "Tick suppressed unless include_stats=true");
}
#[test]
fn eve_tick_included_when_enabled() {
let mut buf = Vec::new();
let mut opts = EveOptions::default();
opts.include_stats = true;
let mut w = EveJsonWriter::with_options(&mut buf, opts);
let mut stats = FlowStats::default();
stats.packets_initiator = 3;
let ev: FlowEvent<FiveTupleKey> = FlowEvent::Tick {
key: key(),
stats,
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
let lines = parse_lines(&buf);
assert_eq!(lines[0]["event_type"], "stats");
assert_eq!(lines[0]["stats"]["pkts_toserver"], 3);
}
#[test]
fn eve_in_iface_present_when_set() {
let mut buf = Vec::new();
let mut opts = EveOptions::default();
opts.in_iface = "eth0".to_string();
let mut w = EveJsonWriter::with_options(&mut buf, opts);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key: key(),
kind: AnomalyKind::RetransmittedSegment {
side: FlowSide::Initiator,
count: 1,
},
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
let lines = parse_lines(&buf);
assert_eq!(lines[0]["in_iface"], "eth0");
}
#[test]
fn eve_every_line_is_valid_json() {
let mut buf = Vec::new();
let mut w = EveJsonWriter::new(&mut buf);
let events: Vec<FlowEvent<FiveTupleKey>> = vec![
FlowEvent::FlowAnomaly {
key: key(),
kind: AnomalyKind::OutOfOrderSegment {
side: FlowSide::Initiator,
count: 1,
},
ts: Timestamp::new(1_700_000_000, 0),
},
FlowEvent::Ended {
key: key(),
reason: EndReason::Fin,
stats: FlowStats::default(),
history: HistoryString::new(),
l4: Some(L4Proto::Tcp),
},
FlowEvent::TrackerAnomaly {
kind: AnomalyKind::FlowTableEvictionPressure {
evicted_in_tick: 1,
evicted_total: 42,
},
ts: Timestamp::new(1_700_000_000, 0),
},
];
for ev in &events {
w.write_event(ev).unwrap();
}
let lines = parse_lines(&buf);
assert_eq!(lines.len(), 3);
}
#[test]
fn eve_severity_numeric_override_can_invert() {
fn invert(s: flowscope::event::Severity) -> u8 {
match s {
flowscope::event::Severity::Critical => 4,
flowscope::event::Severity::Error => 3,
flowscope::event::Severity::Warning => 2,
flowscope::event::Severity::Info => 1,
_ => 0,
}
}
let mut buf = Vec::new();
let mut opts = EveOptions::default();
opts.severity_numeric = invert;
let mut w = EveJsonWriter::with_options(&mut buf, opts);
let ev: FlowEvent<FiveTupleKey> = FlowEvent::FlowAnomaly {
key: key(),
kind: AnomalyKind::OutOfOrderSegment {
side: FlowSide::Initiator,
count: 1,
},
ts: Timestamp::new(1_700_000_000, 0),
};
w.write_event(&ev).unwrap();
let lines = parse_lines(&buf);
assert_eq!(lines[0]["severity"], 1);
}