From 1d6675d00a87afc12a5ff284843559b895db2a5c Mon Sep 17 00:00:00 2001 From: zuub-don Date: Sat, 30 May 2026 22:41:31 -0700 Subject: [PATCH] Add CEF and LEEF parsers (SIEM event formats) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two parsers in ax-normalize for the ArcSight/QRadar SIEM wire formats, each one row per event. The category fields (signatureId/name for CEF, eventId for LEEF) and severity are what dist.chi2 reads as a signature/category mix shift, and a value never seen in the baseline surfaces as a new category automatically. CefParser: 7 pipe-delimited header fields with \| / \\ escaping, then a space-separated extension whose values may themselves contain spaces — a new key begins only at a space-preceded ident=, with \= / \\ / \n value escaping. Severity is type-inferred (numeric or named level); extension values inferred. LeefParser: 5 header fields; LEEF 2.0 adds an explicit delimiter field (a char or xHH hex), LEEF 1.0 uses a tab. Extension is plain key=value pairs. Both detected by their CEF: / LEEF: prefix (STRONG); claim .cef / .leef. A line missing the prefix or with too few header fields is a clean Parse error. Mutation gate: 0 surviving mutants on the new file. The CEF extension scan's i = j + 1 advance has one documented-equivalent variant (+ -> *, i = j skips the '=' next iteration → identical keys); the + -> - variant re-detects forever and is caught as a timeout. Closes #9 Co-Authored-By: Claude Opus 4.8 (1M context) --- .cargo/mutants.toml | 8 + crates/ax-normalize/src/parser.rs | 4 + crates/ax-normalize/src/parsers/cef.rs | 473 +++++++++++++++++++++++++ crates/ax-normalize/src/parsers/mod.rs | 4 + 4 files changed, 489 insertions(+) create mode 100644 crates/ax-normalize/src/parsers/cef.rs diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml index 943f974..f547854 100644 --- a/.cargo/mutants.toml +++ b/.cargo/mutants.toml @@ -48,6 +48,14 @@ exclude_re = [ # be the arg-max once any earlier split has been seen. Equivalent. "replace < with <= in cusum_changepoint", + # CEF extension scan advances `i = j + 1` past the `=` at index j after + # recording a key. The `+`→`*` variant sets `i = j` (the `=` itself); since + # `=` is not an identifier char, the very next iteration skips it and lands on + # j+1 — the identical key set, just one extra no-op step. Equivalent. (The + # `+`→`-` variant is NOT equivalent: it re-detects a 1-char key forever and is + # caught as a timeout by the `a=1 b=2` test.) + 'replace \+ with \* in parse_cef_extension', + # `naive_one_pass_variance` is a deliberately *inaccurate* reference baseline # (the textbook one-pass formula), kept only to demonstrate the precision gap # against our compensated reductions. Its sole contract — "less accurate than diff --git a/crates/ax-normalize/src/parser.rs b/crates/ax-normalize/src/parser.rs index 071d153..4beaf8c 100644 --- a/crates/ax-normalize/src/parser.rs +++ b/crates/ax-normalize/src/parser.rs @@ -219,6 +219,8 @@ mod tests { "logfmt", "accesslog", "syslog", + "cef", + "leef", "prometheus", "json", "yaml", @@ -236,6 +238,8 @@ mod tests { "logfmt", "accesslog", "syslog", + "cef", + "leef", "prometheus", "json", "yaml", diff --git a/crates/ax-normalize/src/parsers/cef.rs b/crates/ax-normalize/src/parsers/cef.rs new file mode 100644 index 0000000..5cce65f --- /dev/null +++ b/crates/ax-normalize/src/parsers/cef.rs @@ -0,0 +1,473 @@ +//! CEF and LEEF parsers — ArcSight / QRadar SIEM event formats. +//! +//! Both are pipe-delimited headers followed by a `key=value` extension. One row +//! per event; the header's category fields (`signatureId`/`name` for CEF, +//! `eventId` for LEEF) and `severity` are exactly what `dist.chi2` reads as a +//! signature/category mix shift, and a value never seen in the baseline surfaces +//! as a new category automatically. +//! +//! - **CEF** (`CEF:Version|Vendor|Product|Version|SignatureID|Name|Severity|ext`): +//! 7 header fields (with `\|` / `\\` escaping) then a space-separated extension +//! whose values may contain spaces — split at ` key=` boundaries, with +//! `\=` / `\\` / `\n` value escaping. +//! - **LEEF** (`LEEF:Version|Vendor|Product|Version|EventID|[Delimiter|]ext`): +//! 5 header fields; LEEF 2.0 adds an explicit delimiter field (a char or `xHH` +//! hex), LEEF 1.0 uses a tab. The extension is plain `key=value` pairs. + +use crate::infer; +use crate::parser::{Confidence, FormatParser, STRONG}; +use crate::table::TableBuilder; +use ax_core::{AxError, Column, Value}; +use std::collections::BTreeMap; + +fn is_ident(c: char) -> bool { + c.is_ascii_alphanumeric() +} + +/// Decodes the SIEM backslash escapes: `\n`/`\r`/`\t` to whitespace, and any +/// other `\x` (e.g. `\|`, `\=`, `\\`) to the literal `x`. +fn unescape(s: &str) -> String { + let mut out = String::new(); + let mut chars = s.chars(); + while let Some(c) = chars.next() { + if c == '\\' { + match chars.next() { + Some('n') => out.push('\n'), + Some('r') => out.push('\r'), + Some('t') => out.push('\t'), + Some(other) => out.push(other), + None => out.push('\\'), + } + } else { + out.push(c); + } + } + out +} + +/// Splits on `|` that is not backslash-escaped, into at most `max` fields (the +/// last absorbs any remaining pipes). Escape pairs are preserved for [`unescape`]. +fn split_unescaped_pipe(s: &str, max: usize) -> Vec { + let mut fields = Vec::new(); + let mut cur = String::new(); + let mut chars = s.chars().peekable(); + while let Some(c) = chars.next() { + if c == '\\' { + cur.push('\\'); + if let Some(next) = chars.next() { + cur.push(next); + } + } else if c == '|' && fields.len() < max - 1 { + fields.push(std::mem::take(&mut cur)); + } else { + cur.push(c); + } + } + fields.push(cur); + fields +} + +/// Parses a CEF extension (`key=value key2=value2 ...`) where a value may itself +/// contain spaces — a new key begins only at a space-preceded `ident=`. Values +/// are unescaped. +fn parse_cef_extension(ext: &str) -> Vec<(String, String)> { + let chars: Vec = ext.chars().collect(); + let n = chars.len(); + // Locate each key: an identifier run, ending in `=`, at start or after a space. + let mut keys: Vec<(usize, usize)> = Vec::new(); // (key_start, eq_index) + let mut i = 0; + while i < n { + if (i == 0 || chars[i - 1] == ' ') && is_ident(chars[i]) { + let mut j = i; + while j < n && is_ident(chars[j]) { + j += 1; + } + if j < n && chars[j] == '=' { + keys.push((i, j)); + i = j + 1; + continue; + } + } + i += 1; + } + let mut pairs = Vec::new(); + for (idx, &(key_start, eq)) in keys.iter().enumerate() { + let key: String = chars[key_start..eq].iter().collect(); + let value_end = keys.get(idx + 1).map_or(n, |&(next_start, _)| next_start); + let raw: String = chars[eq + 1..value_end].iter().collect(); + pairs.push((key, unescape(raw.trim_end()))); + } + pairs +} + +/// Resolves a LEEF 2.0 delimiter spec: a literal char, or `xHH` / `\xHH` hex. +/// Defaults to tab (the LEEF 1.0 separator). +fn leef_delimiter(spec: &str) -> char { + if let Some(hex) = spec.strip_prefix('x').or_else(|| spec.strip_prefix("\\x")) { + if let Ok(byte) = u8::from_str_radix(hex, 16) { + return byte as char; + } + } + spec.chars().next().unwrap_or('\t') +} + +// ----------------------------------------------------------------- CEF -------- + +#[derive(Debug, Default, Clone)] +pub struct CefParser; + +const CEF_HEADER: [&str; 7] = [ + "cefVersion", + "deviceVendor", + "deviceProduct", + "deviceVersion", + "signatureId", + "name", + "severity", +]; + +impl CefParser { + fn err(&self, msg: impl std::fmt::Display) -> AxError { + AxError::Parse { + format: self.id().to_string(), + message: msg.to_string(), + } + } +} + +impl FormatParser for CefParser { + fn id(&self) -> &'static str { + "cef" + } + fn extensions(&self) -> &'static [&'static str] { + &["cef"] + } + fn sniff(&self, bytes: &[u8]) -> Option { + let text = std::str::from_utf8(bytes).ok()?; + let line = text.lines().find(|l| !l.trim().is_empty())?; + line.starts_with("CEF:").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 rest = line + .strip_prefix("CEF:") + .ok_or_else(|| self.err("not a CEF line: missing 'CEF:' prefix"))?; + let fields = split_unescaped_pipe(rest, 8); + if fields.len() < CEF_HEADER.len() { + return Err(self.err("CEF header requires 7 pipe-delimited fields")); + } + let mut row: BTreeMap = BTreeMap::new(); + for (name, raw) in CEF_HEADER.iter().zip(&fields) { + let decoded = unescape(raw); + // Severity is the analyzable numeric (or a named level); the rest + // are categorical identifiers kept verbatim. + let cell = if *name == "severity" { + infer::infer_scalar(&decoded) + } else { + Value::Str(decoded) + }; + row.insert((*name).to_string(), cell); + } + if let Some(ext) = fields.get(CEF_HEADER.len()) { + for (key, value) in parse_cef_extension(ext) { + row.insert(key, infer::infer_scalar(&value)); + } + } + builder.push_row(row); + } + Ok(builder.finish()) + } +} + +// ---------------------------------------------------------------- LEEF -------- + +#[derive(Debug, Default, Clone)] +pub struct LeefParser; + +const LEEF_HEADER: [&str; 5] = [ + "leefVersion", + "vendor", + "product", + "productVersion", + "eventId", +]; + +impl LeefParser { + fn err(&self, msg: impl std::fmt::Display) -> AxError { + AxError::Parse { + format: self.id().to_string(), + message: msg.to_string(), + } + } +} + +impl FormatParser for LeefParser { + fn id(&self) -> &'static str { + "leef" + } + fn extensions(&self) -> &'static [&'static str] { + &["leef"] + } + fn sniff(&self, bytes: &[u8]) -> Option { + let text = std::str::from_utf8(bytes).ok()?; + let line = text.lines().find(|l| !l.trim().is_empty())?; + line.starts_with("LEEF:").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 rest = line + .strip_prefix("LEEF:") + .ok_or_else(|| self.err("not a LEEF line: missing 'LEEF:' prefix"))?; + // LEEF 2.0 inserts a delimiter field between the header and extension. + let version = rest.split('|').next().unwrap_or(""); + let is_v2 = version.starts_with('2'); + let header_count = LEEF_HEADER.len() + usize::from(is_v2); + let parts: Vec<&str> = rest.splitn(header_count + 1, '|').collect(); + if parts.len() < LEEF_HEADER.len() { + return Err(self.err("LEEF header requires at least 5 fields")); + } + let mut row: BTreeMap = BTreeMap::new(); + for (name, value) in LEEF_HEADER.iter().zip(&parts) { + row.insert((*name).to_string(), Value::Str((*value).to_string())); + } + let delimiter = if is_v2 { + parts + .get(LEEF_HEADER.len()) + .map_or('\t', |s| leef_delimiter(s)) + } else { + '\t' + }; + if let Some(ext) = parts.get(header_count) { + for token in ext.split(delimiter) { + if let Some((key, value)) = token.split_once('=') { + if !key.is_empty() { + row.insert(key.to_string(), infer::infer_scalar(value)); + } + } + } + } + builder.push_row(row); + } + Ok(builder.finish()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use ax_core::ColType; + + fn col<'a>(cols: &'a [Column], name: &str) -> &'a Column { + cols.iter() + .find(|c| c.name == name) + .unwrap_or_else(|| panic!("missing column {name}")) + } + + // -------------------------------------------------------- helpers -------- + + #[test] + fn unescape_decodes_siem_escapes() { + assert_eq!(unescape(r"a\|b"), "a|b"); + assert_eq!(unescape(r"a\=b"), "a=b"); + assert_eq!(unescape(r"a\\b"), r"a\b"); + assert_eq!(unescape(r"a\nb"), "a\nb"); + assert_eq!(unescape("plain"), "plain"); + } + + #[test] + fn split_unescaped_pipe_keeps_escaped_and_extra() { + // Escaped pipe stays in its field; once 8 fields are reached the last + // absorbs any further pipes (9 segments here → 8th keeps "i|j"). + let f = split_unescaped_pipe(r"a\|b|c|d|e|f|g|h|i|j", 8); + assert_eq!(f.len(), 8); + assert_eq!(f[0], r"a\|b", "escaped pipe is not a separator"); + assert_eq!(f[7], "i|j", "extension field absorbs extra pipes"); + } + + #[test] + fn parse_cef_extension_handles_spaces_and_escapes() { + let pairs = parse_cef_extension(r"src=10.0.0.1 msg=worm was stopped spt=1232 note=a\=b"); + assert_eq!( + pairs, + vec![ + ("src".into(), "10.0.0.1".into()), + ("msg".into(), "worm was stopped".into()), // value with spaces + ("spt".into(), "1232".into()), + ("note".into(), "a=b".into()), // escaped '=' + ] + ); + } + + #[test] + fn parse_cef_extension_only_breaks_at_space_preceded_keys() { + // A `=` that is not space-preceded stays inside the value (a new key + // begins ONLY after a space). And single-char keys advance correctly. + assert_eq!( + parse_cef_extension("k=ab=cd"), + vec![("k".into(), "ab=cd".into())] + ); + assert_eq!( + parse_cef_extension("a=1 b=2"), + vec![("a".into(), "1".into()), ("b".into(), "2".into())] + ); + } + + #[test] + fn leef_delimiter_resolves_char_hex_and_default() { + assert_eq!(leef_delimiter("^"), '^'); + assert_eq!(leef_delimiter("x09"), '\t'); + assert_eq!(leef_delimiter(r"\x09"), '\t'); + assert_eq!(leef_delimiter(""), '\t'); // empty → tab default + } + + // ----------------------------------------------------------- CEF -------- + + const CEF: &str = concat!( + r"CEF:0|Security|threatmanager|1.0|100|worm stopped|10|src=10.0.0.1 spt=1232 msg=took action", + "\n", + r"CEF:0|Security|threatmanager|1.0|200|port scan|3|src=10.0.0.9 dst=2.1.2.2", + "\n", + ); + + fn cef(s: &str) -> Vec { + CefParser.parse("-", s.as_bytes()).unwrap() + } + + #[test] + fn cef_header_fields() { + let cols = cef(CEF); + assert_eq!( + col(&cols, "deviceProduct").cells[0], + Value::Str("threatmanager".into()) + ); + assert_eq!(col(&cols, "signatureId").cells[0], Value::Str("100".into())); + assert_eq!(col(&cols, "name").cells[1], Value::Str("port scan".into())); + let sev = col(&cols, "severity"); + assert_eq!(sev.ty, ColType::Int, "severity is the analyzable numeric"); + assert_eq!(sev.cells, vec![Value::Int(10), Value::Int(3)]); + } + + #[test] + fn cef_extension_fields_typed_and_padded() { + let cols = cef(CEF); + assert_eq!(col(&cols, "src").cells[0], Value::Str("10.0.0.1".into())); + assert_eq!(col(&cols, "spt").cells[0], Value::Int(1232)); // port → int + assert_eq!(col(&cols, "msg").cells[0], Value::Str("took action".into())); + // dst only on the second event; spt/msg only on the first. + assert_eq!(col(&cols, "dst").cells[0], Value::Null); + assert_eq!(col(&cols, "spt").cells[1], Value::Null); + } + + #[test] + fn cef_escaped_pipe_in_header() { + let cols = cef(r"CEF:0|Sec\|ops|prod|1|1|n|5|"); + assert_eq!( + col(&cols, "deviceVendor").cells[0], + Value::Str("Sec|ops".into()) + ); + } + + #[test] + fn cef_without_extension() { + let cols = cef("CEF:0|v|p|1.0|42|evt|7\n"); // 7 fields, no extension + assert_eq!(col(&cols, "signatureId").cells[0], Value::Str("42".into())); + assert_eq!(col(&cols, "severity").cells[0], Value::Int(7)); + } + + #[test] + fn cef_malformed_too_few_fields_errors() { + assert!(matches!( + CefParser.parse("-", b"CEF:0|only|three\n"), + Err(AxError::Parse { .. }) + )); + assert!(matches!( + CefParser.parse("-", b"not a cef line\n"), + Err(AxError::Parse { .. }) + )); + } + + #[test] + fn cef_sniff_and_resolution() { + assert_eq!(CefParser.sniff(CEF.as_bytes()), Some(STRONG)); + assert_eq!(CefParser.sniff(b"LEEF:1.0|v|p|1|x|"), None); + assert_eq!(CefParser.sniff(b"a,b,c\n1,2,3"), None); + assert_eq!(CefParser.extensions(), &["cef"]); + let reg = crate::parser::ParserRegistry::default(); + assert_eq!(reg.resolve("e.cef", b"x").unwrap().id(), "cef"); + assert_eq!(reg.resolve("-", CEF.as_bytes()).unwrap().id(), "cef"); + } + + // ---------------------------------------------------------- LEEF -------- + + #[test] + fn leef_v1_tab_extension() { + let line = "LEEF:1.0|Lancope|StealthWatch|1.0|41|src=192.0.2.0\tdst=172.50.123.1\tsev=5\n"; + let cols = LeefParser.parse("-", line.as_bytes()).unwrap(); + assert_eq!(col(&cols, "leefVersion").cells[0], Value::Str("1.0".into())); + assert_eq!(col(&cols, "vendor").cells[0], Value::Str("Lancope".into())); + assert_eq!(col(&cols, "eventId").cells[0], Value::Str("41".into())); + assert_eq!(col(&cols, "src").cells[0], Value::Str("192.0.2.0".into())); + assert_eq!(col(&cols, "sev").cells[0], Value::Int(5)); + } + + #[test] + fn leef_header_only_no_extension() { + // Exactly 5 fields (no extension) is valid LEEF 1.0 — pins the `< 5` + // header-count boundary (must not reject a 5-field header). + let cols = LeefParser.parse("-", b"LEEF:1.0|Acme|Tool|2|77").unwrap(); + assert_eq!(col(&cols, "eventId").cells[0], Value::Str("77".into())); + assert_eq!(col(&cols, "vendor").cells[0], Value::Str("Acme".into())); + } + + #[test] + fn leef_v2_explicit_delimiter() { + // LEEF 2.0 with a '^' delimiter field between header and extension. + let line = "LEEF:2.0|Vendor|Product|2.5|1001|^|src=10.0.0.1^dst=10.0.0.2^spt=22\n"; + let cols = LeefParser.parse("-", line.as_bytes()).unwrap(); + assert_eq!(col(&cols, "eventId").cells[0], Value::Str("1001".into())); + assert_eq!(col(&cols, "src").cells[0], Value::Str("10.0.0.1".into())); + assert_eq!(col(&cols, "spt").cells[0], Value::Int(22)); + // The delimiter field itself is not a data column. + assert!(cols.iter().all(|c| c.name != "^")); + } + + #[test] + fn leef_v2_hex_delimiter() { + // x09 = tab delimiter. + let line = "LEEF:2.0|V|P|1|99|x09|a=1\tb=2\n"; + let cols = LeefParser.parse("-", line.as_bytes()).unwrap(); + assert_eq!(col(&cols, "a").cells[0], Value::Int(1)); + assert_eq!(col(&cols, "b").cells[0], Value::Int(2)); + } + + #[test] + fn leef_malformed_and_sniff() { + assert!(matches!( + LeefParser.parse("-", b"LEEF:1.0|onlytwo\n"), + Err(AxError::Parse { .. }) + )); + assert!(matches!( + LeefParser.parse("-", b"not leef\n"), + Err(AxError::Parse { .. }) + )); + assert_eq!(LeefParser.sniff(b"LEEF:1.0|v|p|1|x|a=1"), Some(STRONG)); + assert_eq!(LeefParser.sniff(b"CEF:0|v|p|1|1|n|5|"), None); + assert_eq!(LeefParser.extensions(), &["leef"]); + let reg = crate::parser::ParserRegistry::default(); + assert_eq!(reg.resolve("e.leef", b"x").unwrap().id(), "leef"); + assert_eq!( + reg.resolve("-", b"LEEF:1.0|v|p|1|x|a=1\n").unwrap().id(), + "leef" + ); + } +} diff --git a/crates/ax-normalize/src/parsers/mod.rs b/crates/ax-normalize/src/parsers/mod.rs index 0d7f8a7..442dacc 100644 --- a/crates/ax-normalize/src/parsers/mod.rs +++ b/crates/ax-normalize/src/parsers/mod.rs @@ -5,6 +5,7 @@ use crate::parser::ParserRegistry; pub mod accesslog; +pub mod cef; pub mod delimited; pub mod journal; pub mod json; @@ -21,6 +22,7 @@ pub mod zeek; pub mod columnar; pub use accesslog::AccessLogParser; +pub use cef::{CefParser, LeefParser}; pub use delimited::{CsvParser, TsvParser}; pub use journal::JournalParser; pub use json::JsonParser; @@ -53,6 +55,8 @@ pub fn default_registry() -> ParserRegistry { r.register(Box::new(LogfmtParser)); r.register(Box::new(AccessLogParser)); r.register(Box::new(SyslogParser)); + r.register(Box::new(CefParser)); + r.register(Box::new(LeefParser)); r.register(Box::new(PrometheusParser)); r.register(Box::new(JsonParser)); r.register(Box::new(YamlParser));