From 966d08cfb9da105e0172607978eeb38ef440a83b Mon Sep 17 00:00:00 2001 From: zuub-don Date: Sat, 30 May 2026 22:58:52 -0700 Subject: [PATCH] Add Suricata/Zeek EVE JSON parser (IDS alerts) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New EveParser in ax-normalize. EVE is NDJSON with each event tagged by an event_type and timestamp, and the interesting fields one level down in a per-type object. Generic NDJSON would stringify those nested objects; this parser flattens them into dotted columns (alert.category, alert.severity, dns.rrname, ...) so a field like alert.category is its own column — exactly what dist.chi2 --baseline reads as alert-type drift (quiet vs incident window), with a brand-new alert class surfacing as a category never seen in the baseline. Arrays are kept as canonical JSON (not exploded); scalars typed via json_to_value. Detected by the event_type (string) + timestamp signature and registered before NDJSON so an EVE stream claims itself before the generic NDJSON shape. Claims no extension (EVE is generically eve.json, owned by the JSON parser). Non-object / non-JSON lines are a clean Parse error. Mutation gate: 0 surviving mutants on the new file. The extensions() -> empty-slice mutant is a documented equivalent (same as Zeek/Journal). Closes #10 Co-Authored-By: Claude Opus 4.8 (1M context) --- .cargo/mutants.toml | 4 + crates/ax-normalize/src/parser.rs | 2 + crates/ax-normalize/src/parsers/eve.rs | 229 +++++++++++++++++++++++++ crates/ax-normalize/src/parsers/mod.rs | 7 +- 4 files changed, 240 insertions(+), 2 deletions(-) create mode 100644 crates/ax-normalize/src/parsers/eve.rs diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml index f547854..a37a687 100644 --- a/.cargo/mutants.toml +++ b/.cargo/mutants.toml @@ -20,6 +20,10 @@ exclude_re = [ # behavior — no input can observe the difference. (Same case as Zeek above.) 'JournalParser>::extensions.*Vec::new', + # EveParser also claims no extension (returns `&[]`) — same empty-slice + # equivalence as Zeek/Journal above. + 'EveParser>::extensions.*Vec::new', + # Parser sniff confidences are registered in strictly descending order # (MAGIC > STRONG > TEXT > FALLBACK), so the highest-confidence match is # always unique — no two firing parsers ever tie. The `>` → `==` / `>=` diff --git a/crates/ax-normalize/src/parser.rs b/crates/ax-normalize/src/parser.rs index 4beaf8c..daae1ee 100644 --- a/crates/ax-normalize/src/parser.rs +++ b/crates/ax-normalize/src/parser.rs @@ -213,6 +213,7 @@ mod tests { "parquet", "arrow", "otlp", + "eve", "journal", "ndjson", "zeek", @@ -232,6 +233,7 @@ mod tests { #[cfg(not(feature = "polars"))] let expected = vec![ "otlp", + "eve", "journal", "ndjson", "zeek", diff --git a/crates/ax-normalize/src/parsers/eve.rs b/crates/ax-normalize/src/parsers/eve.rs new file mode 100644 index 0000000..39ad325 --- /dev/null +++ b/crates/ax-normalize/src/parsers/eve.rs @@ -0,0 +1,229 @@ +//! Suricata/Zeek EVE JSON parser — the standard IDS alert stream. +//! +//! EVE is NDJSON (one event per line), every event tagged with an `event_type` +//! (`alert`, `dns`, `flow`, `http`, …) and a `timestamp`. The interesting fields +//! live one level down in a per-type object (`alert.signature`, `alert.category`, +//! `alert.severity`, `dns.rrname`, …). Generic NDJSON would stringify those +//! nested objects; this parser **flattens them into dotted columns** so a value +//! like `alert.category` is its own column — exactly what `dist.chi2 --baseline` +//! reads as alert-type drift (quiet vs incident window), with a brand-new alert +//! class surfacing as a category never seen in the baseline. +//! +//! Arrays are kept as their canonical JSON string (not exploded). Detected by the +//! `event_type` + `timestamp` signature; claims no extension (EVE is generically +//! `eve.json`, owned by the JSON parser — pipe it on stdin for EVE-aware +//! flattening). + +use crate::infer; +use crate::parser::{Confidence, FormatParser, STRONG}; +use crate::table::TableBuilder; +use ax_core::{AxError, Column, Value}; +use serde_json::Value as J; +use std::collections::BTreeMap; + +#[derive(Debug, Default, Clone)] +pub struct EveParser; + +fn looks_like_eve(obj: &serde_json::Map) -> bool { + obj.get("event_type").is_some_and(J::is_string) && obj.contains_key("timestamp") +} + +/// Flattens nested objects into dotted keys (`alert.category`); scalars and +/// arrays are leaves (arrays kept as canonical JSON via [`infer::json_to_value`]). +fn flatten(prefix: &str, value: &J, row: &mut BTreeMap) { + match value { + J::Object(map) => { + for (key, val) in map { + let dotted = if prefix.is_empty() { + key.clone() + } else { + format!("{prefix}.{key}") + }; + flatten(&dotted, val, row); + } + } + leaf => { + row.insert(prefix.to_string(), infer::json_to_value(leaf)); + } + } +} + +impl EveParser { + fn err(&self, msg: impl std::fmt::Display) -> AxError { + AxError::Parse { + format: self.id().to_string(), + message: msg.to_string(), + } + } +} + +impl FormatParser for EveParser { + fn id(&self) -> &'static str { + "eve" + } + fn extensions(&self) -> &'static [&'static str] { + &[] + } + fn sniff(&self, bytes: &[u8]) -> Option { + let text = std::str::from_utf8(bytes).ok()?; + let line = text.lines().find(|l| !l.trim().is_empty())?; + let value: J = serde_json::from_str(line).ok()?; + value + .as_object() + .is_some_and(looks_like_eve) + .then_some(STRONG) + } + fn parse(&self, _source: &str, bytes: &[u8]) -> Result, AxError> { + let text = std::str::from_utf8(bytes).map_err(|e| self.err(e))?; + let mut builder = TableBuilder::new(); + for line in text.lines() { + if line.trim().is_empty() { + continue; + } + let value: J = serde_json::from_str(line).map_err(|e| self.err(e))?; + if !value.is_object() { + return Err(self.err("EVE event is not a JSON object")); + } + let mut row: BTreeMap = BTreeMap::new(); + flatten("", &value, &mut row); + builder.push_row(row); + } + Ok(builder.finish()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use ax_core::ColType; + + const EVE: &str = concat!( + r#"{"timestamp":"2017-01-01T00:00:01.0+0000","flow_id":12,"event_type":"alert","src_ip":"1.2.3.4","src_port":1234,"dest_ip":"5.6.7.8","dest_port":80,"proto":"TCP","alert":{"signature_id":2010935,"signature":"ET POLICY external IP","category":"Potential Corporate Privacy Violation","severity":1},"metadata":["a","b"]}"#, + "\n", + r#"{"timestamp":"2017-01-01T00:00:02.0+0000","flow_id":13,"event_type":"dns","src_ip":"1.2.3.4","proto":"UDP","dns":{"type":"query","rrname":"example.com"}}"#, + "\n", + ); + + fn parse(s: &str) -> Vec { + EveParser.parse("-", s.as_bytes()).unwrap() + } + fn col<'a>(cols: &'a [Column], name: &str) -> &'a Column { + cols.iter() + .find(|c| c.name == name) + .unwrap_or_else(|| panic!("missing column {name}")) + } + + #[test] + fn top_level_fields_are_typed() { + let cols = parse(EVE); + assert_eq!( + col(&cols, "event_type").cells[0], + Value::Str("alert".into()) + ); + assert_eq!(col(&cols, "src_port").ty, ColType::Int); + assert_eq!(col(&cols, "src_port").cells[0], Value::Int(1234)); + assert_eq!(col(&cols, "dest_ip").cells[0], Value::Str("5.6.7.8".into())); + assert_eq!(col(&cols, "flow_id").cells[1], Value::Int(13)); + } + + #[test] + fn nested_alert_object_is_flattened_to_dotted_columns() { + let cols = parse(EVE); + assert_eq!( + col(&cols, "alert.category").cells[0], + Value::Str("Potential Corporate Privacy Violation".into()) + ); + assert_eq!(col(&cols, "alert.severity").ty, ColType::Int); + assert_eq!(col(&cols, "alert.severity").cells[0], Value::Int(1)); + assert_eq!( + col(&cols, "alert.signature_id").cells[0], + Value::Int(2010935) + ); + // The dns event has no alert.* fields → padded null. + assert_eq!(col(&cols, "alert.category").cells[1], Value::Null); + } + + #[test] + fn second_event_type_flattens_its_own_object() { + let cols = parse(EVE); + assert_eq!(col(&cols, "event_type").cells[1], Value::Str("dns".into())); + assert_eq!( + col(&cols, "dns.rrname").cells[1], + Value::Str("example.com".into()) + ); + assert_eq!(col(&cols, "dns.rrname").cells[0], Value::Null); // alert row + } + + #[test] + fn arrays_are_kept_as_canonical_json() { + let cols = parse(EVE); + assert_eq!( + col(&cols, "metadata").cells[0], + Value::Str("[\"a\",\"b\"]".into()) + ); + } + + #[test] + fn flatten_units() { + let mut row = BTreeMap::new(); + flatten( + "", + &serde_json::json!({"a": {"b": {"c": 5}}, "d": 2}), + &mut row, + ); + assert_eq!(row.get("a.b.c"), Some(&Value::Int(5))); // deep nesting + assert_eq!(row.get("d"), Some(&Value::Int(2))); + assert_eq!(row.len(), 2); + } + + #[test] + fn malformed_events_error() { + assert!(matches!( + EveParser.parse("-", b"not json\n"), + Err(AxError::Parse { .. }) + )); + assert!(matches!( + EveParser.parse("-", b"[1,2,3]\n"), // valid JSON, not an object + Err(AxError::Parse { .. }) + )); + } + + #[test] + fn sniff_keys_on_event_type_and_timestamp() { + assert_eq!(EveParser.sniff(EVE.as_bytes()), Some(STRONG)); + // event_type must be a string AND timestamp present. + assert_eq!( + EveParser.sniff(br#"{"event_type":"alert","timestamp":"t"}"#), + Some(STRONG) + ); + assert_eq!( + EveParser.sniff(br#"{"event_type":"alert"}"#), + None, + "no timestamp" + ); + assert_eq!( + EveParser.sniff(br#"{"event_type":5,"timestamp":"t"}"#), + None, + "event_type not a string" + ); + // Generic NDJSON without the EVE signature is not EVE. + assert_eq!(EveParser.sniff(b"{\"a\":1}\n{\"a\":2}\n"), None); + assert_eq!(EveParser.sniff(b"a,b,c\n1,2,3"), None); // not JSON + } + + #[test] + fn claims_no_extension() { + assert!(EveParser.extensions().is_empty()); + } + + #[test] + fn resolves_eve_over_ndjson_by_content() { + let reg = crate::parser::ParserRegistry::default(); + assert_eq!(reg.resolve("-", EVE.as_bytes()).unwrap().id(), "eve"); + // Generic NDJSON (no EVE signature) is still NDJSON. + assert_eq!( + reg.resolve("-", b"{\"a\":1}\n{\"a\":2}\n").unwrap().id(), + "ndjson" + ); + } +} diff --git a/crates/ax-normalize/src/parsers/mod.rs b/crates/ax-normalize/src/parsers/mod.rs index 442dacc..194934e 100644 --- a/crates/ax-normalize/src/parsers/mod.rs +++ b/crates/ax-normalize/src/parsers/mod.rs @@ -7,6 +7,7 @@ use crate::parser::ParserRegistry; pub mod accesslog; pub mod cef; pub mod delimited; +pub mod eve; pub mod journal; pub mod json; pub mod logfmt; @@ -24,6 +25,7 @@ pub mod columnar; pub use accesslog::AccessLogParser; pub use cef::{CefParser, LeefParser}; pub use delimited::{CsvParser, TsvParser}; +pub use eve::EveParser; pub use journal::JournalParser; pub use json::JsonParser; pub use logfmt::LogfmtParser; @@ -47,8 +49,9 @@ pub fn default_registry() -> ParserRegistry { // OTLP before NDJSON: a compact single-object OTLP doc must win the // `resourceSpans` signature before any JSON-line heuristic sees it. r.register(Box::new(OtlpParser)); - // Journal before NDJSON: a journald export is NDJSON, so its trusted-field - // signature must claim it before the generic NDJSON shape does. + // EVE and Journal before NDJSON: both are NDJSON dialects, so their content + // signatures must claim them before the generic NDJSON shape does. + r.register(Box::new(EveParser)); r.register(Box::new(JournalParser)); r.register(Box::new(NdjsonParser)); r.register(Box::new(ZeekParser));