From 114d0f06d5e1f4061d7b09f5a51cdad5e845976c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 23 Mar 2026 23:33:36 +0000 Subject: [PATCH 1/4] Initial plan From 64a8e864ff52dfe0103b4f7517fc7105f5c0c74a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 23 Mar 2026 23:52:29 +0000 Subject: [PATCH 2/4] Implement Sigma correlation rule matching with sliding-window engine Co-authored-by: thomaspatzke <1845601+thomaspatzke@users.noreply.github.com> Agent-Logs-Url: https://github.com/SigmaHQ/sigma_engine/sessions/3b33176e-6be6-4dd8-b3cc-152e3e51c186 --- Cargo.lock | 183 ++++++ Cargo.toml | 2 +- src/lib.rs | 14 +- src/processor.rs | 1538 ++++++++++++++++++++++++++++++++++++++++++++-- 4 files changed, 1694 insertions(+), 43 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 09aa1e0..38f06a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,6 +11,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "android_system_properties" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" +dependencies = [ + "libc", +] + [[package]] name = "anyhow" version = "1.0.102" @@ -35,6 +44,22 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" +[[package]] +name = "bumpalo" +version = "3.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb" + +[[package]] +name = "cc" +version = "1.2.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a0dd1ca384932ff3641c8718a02769f1698e7563dc6974ffd03346116310423" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -47,9 +72,17 @@ version = "0.4.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0" dependencies = [ + "iana-time-zone", "num-traits", + "windows-link", ] +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + [[package]] name = "crossbeam-channel" version = "0.5.15" @@ -87,6 +120,12 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" +[[package]] +name = "find-msvc-tools" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" + [[package]] name = "foldhash" version = "0.1.5" @@ -133,6 +172,30 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + [[package]] name = "id-arena" version = "2.3.0" @@ -157,6 +220,16 @@ version = "1.0.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" +[[package]] +name = "js-sys" +version = "0.3.91" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b49715b7073f385ba4bc528e5747d02e66cb39c6146efb66b781f131f0fb399c" +dependencies = [ + "once_cell", + "wasm-bindgen", +] + [[package]] name = "leb128fmt" version = "0.1.0" @@ -288,6 +361,12 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "rustversion" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" + [[package]] name = "ryu" version = "1.0.23" @@ -356,6 +435,12 @@ dependencies = [ "unsafe-libyaml", ] +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + [[package]] name = "sigma_engine" version = "0.1.0" @@ -453,6 +538,51 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "wasm-bindgen" +version = "0.2.114" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6532f9a5c1ece3798cb1c2cfdba640b9b3ba884f5db45973a6f442510a87d38e" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.114" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18a2d50fcf105fb33bb15f00e7a77b772945a2ee45dcf454961fd843e74c18e6" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.114" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03ce4caeaac547cdf713d280eda22a730824dd11e6b8c3ca9e42247b25c631e3" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.114" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75a326b8c223ee17883a4251907455a2431acc2791c98c26279376490c378c16" +dependencies = [ + "unicode-ident", +] + [[package]] name = "wasm-encoder" version = "0.244.0" @@ -487,12 +617,65 @@ dependencies = [ "semver", ] +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.61.2" diff --git a/Cargo.toml b/Cargo.toml index 5ca7625..9ad7587 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,7 +9,7 @@ license = "MIT" serde = { version = "1", features = ["derive"] } serde_yaml = "0.9" thiserror = "2" -chrono = { version = "0.4", default-features = false, features = ["std"] } +chrono = { version = "0.4", default-features = false, features = ["std", "clock"] } base64 = "0.22" serde_json = "1" crossbeam-channel = "0.5" diff --git a/src/lib.rs b/src/lib.rs index 4847f25..f1835c4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -85,7 +85,7 @@ //! ### Processing events with multithreading //! //! ```rust -//! use sigma_engine::{SigmaCollection, SigmaDocument, LogProcessor, LogEvent, LogSource}; +//! use sigma_engine::{SigmaCollection, SigmaDocument, LogProcessor, LogEvent, LogSource, DetectionResult}; //! use std::collections::HashMap; //! //! let yaml = r#" @@ -126,8 +126,11 @@ //! drop(event_tx); // Signal completion //! //! // Receive detections -//! if let Ok(detection) = detection_rx.recv() { -//! println!("Rule matched: {}", detection.rule.title); +//! if let Ok(result) = detection_rx.recv() { +//! match result { +//! DetectionResult::Rule(detection) => println!("Rule matched: {}", detection.rule.title), +//! DetectionResult::Correlation(cd) => println!("Correlation triggered: {}", cd.rule.title), +//! } //! } //! ``` @@ -143,7 +146,10 @@ pub use error::Error; pub use pipeline::*; pub use types::*; pub use matcher::SigmaRuleMatcher; -pub use processor::{LogProcessor, Detection, LogEvent}; +pub use processor::{ + parse_timespan, CorrelationDetection, Detection, DetectionResult, LogEvent, LogProcessor, + ProcessorConfig, +}; // Re-export chrono's NaiveDate for date field access pub use chrono::NaiveDate; diff --git a/src/processor.rs b/src/processor.rs index c4cee0c..2264aca 100644 --- a/src/processor.rs +++ b/src/processor.rs @@ -46,8 +46,12 @@ //! drop(event_tx); //! //! // Receive detections -//! while let Ok(detection) = detection_rx.recv() { -//! println!("Matched: {}", detection.rule.title); +//! while let Ok(result) = detection_rx.recv() { +//! use sigma_engine::DetectionResult; +//! match result { +//! DetectionResult::Rule(d) => println!("Matched: {}", d.rule.title), +//! DetectionResult::Correlation(cd) => println!("Correlation: {}", cd.rule.title), +//! } //! } //! ``` //! @@ -64,15 +68,20 @@ //! processes events from a shared channel and outputs detections to another channel. //! This allows for efficient parallel processing of high-volume log streams. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::Arc; use std::thread; +use std::time::Duration; +use chrono::{DateTime, Utc}; use crossbeam_channel::{bounded, unbounded, Receiver, Sender}; use serde_json::Value as JsonValue; use crate::matcher::SigmaRuleMatcher; -use crate::types::{LogSource, SigmaRule}; +use crate::types::{ + CorrelationCondition, CorrelationType, LogSource, SigmaCorrelationRule, SigmaDocument, + SigmaRule, +}; /// A log event that can be matched against Sigma rules. #[derive(Debug, Clone)] @@ -83,15 +92,21 @@ pub struct LogEvent { pub data: HashMap, /// The raw event string (if available) pub raw: Option, + /// The time at which this event occurred. + pub timestamp: DateTime, } impl LogEvent { /// Create a new log event from structured field-value pairs. + /// + /// The event timestamp is set to the current UTC time. Use + /// [`LogEvent::with_timestamp`] to override it. pub fn from_fields(log_source: LogSource, data: HashMap) -> Self { Self { log_source, data, raw: None, + timestamp: Utc::now(), } } @@ -106,6 +121,7 @@ impl LogEvent { log_source, data, raw: Some(json.to_string()), + timestamp: Utc::now(), }) } @@ -117,6 +133,7 @@ impl LogEvent { log_source, data, raw: Some(text), + timestamp: Utc::now(), } } @@ -129,9 +146,49 @@ impl LogEvent { log_source, data, raw: Some(text.to_string()), + timestamp: Utc::now(), } } + /// Set the timestamp of this event, consuming and returning the modified event. + /// + /// Use this to assign a specific timestamp rather than the default (current UTC time). + /// + /// # Examples + /// + /// ```rust + /// use sigma_engine::{LogEvent, LogSource}; + /// use chrono::{DateTime, Utc, TimeZone}; + /// use std::collections::HashMap; + /// + /// let log_source = LogSource { category: None, product: None, service: None }; + /// let ts: DateTime = Utc.with_ymd_and_hms(2024, 1, 1, 12, 0, 0).unwrap(); + /// let event = LogEvent::from_fields(log_source, HashMap::new()) + /// .with_timestamp(ts); + /// + /// assert_eq!(event.get_time(), ts); + /// ``` + pub fn with_timestamp(mut self, timestamp: DateTime) -> Self { + self.timestamp = timestamp; + self + } + + /// Returns the event's timestamp. + /// + /// # Examples + /// + /// ```rust + /// use sigma_engine::{LogEvent, LogSource}; + /// use std::collections::HashMap; + /// + /// let log_source = LogSource { category: None, product: None, service: None }; + /// let event = LogEvent::from_fields(log_source, HashMap::new()); + /// let _ts = event.get_time(); // returns DateTime + /// ``` + pub fn get_time(&self) -> DateTime { + self.timestamp + } + /// Convert a JSON value to a flat field-value map. fn json_to_fields(value: &JsonValue) -> HashMap { let mut fields = HashMap::new(); @@ -155,7 +212,7 @@ impl LogEvent { JsonValue::Array(arr) => { // Convert arrays to comma-separated strings let values: Vec = arr.iter() - .map(|v| Self::json_value_to_string(v)) + .map(Self::json_value_to_string) .collect(); fields.insert(prefix, values.join(",")); } @@ -186,7 +243,7 @@ impl LogEvent { while chars.peek().is_some() { // Skip whitespace - while chars.peek().map_or(false, |c| c.is_whitespace()) { + while chars.peek().is_some_and(|c| c.is_whitespace()) { chars.next(); } @@ -206,7 +263,7 @@ impl LogEvent { } // Skip whitespace after '=' - while chars.peek().map_or(false, |c| c.is_whitespace()) { + while chars.peek().is_some_and(|c| c.is_whitespace()) { chars.next(); } @@ -271,6 +328,33 @@ pub struct Detection { pub event: LogEvent, } +/// A detection result produced when a correlation rule's condition is satisfied. +/// +/// A correlation detection is triggered when a set of individual rule detections +/// satisfies the correlation rule's condition (e.g. event count threshold, all +/// temporal rules fired, etc.) within the configured time window. +#[derive(Debug, Clone)] +pub struct CorrelationDetection { + /// The correlation rule whose condition was satisfied. + pub rule: Arc, + /// The group-by field values that identify this correlation group. + /// Keys are `group_by` field names; values are the shared field values. + pub group_key: HashMap, + /// All events within the time window that contributed to this detection. + pub contributing_events: Vec, +} + +/// Unified output from the log processor — either a plain rule match or a correlation match. +/// +/// Received from the channel returned by [`LogProcessor::start`]. +#[derive(Debug, Clone)] +pub enum DetectionResult { + /// A standard Sigma detection rule matched an event. + Rule(Detection), + /// A Sigma correlation rule's condition was satisfied. + Correlation(CorrelationDetection), +} + /// Configuration for the log processor. #[derive(Debug, Clone)] pub struct ProcessorConfig { @@ -280,6 +364,10 @@ pub struct ProcessorConfig { pub event_buffer_size: usize, /// Size of the detection output channel buffer (0 = unbounded) pub detection_buffer_size: usize, + /// Maximum age (in seconds) of an event relative to the newest event seen + /// before it is treated as a late-arriving event and discarded from the + /// correlation time window. Defaults to 3600 (one hour). + pub late_event_threshold_secs: u64, } impl Default for ProcessorConfig { @@ -289,23 +377,31 @@ impl Default for ProcessorConfig { num_threads: if num_cpus > 1 { num_cpus - 1 } else { 1 }, event_buffer_size: 1000, detection_buffer_size: 1000, + late_event_threshold_secs: 3600, } } } -/// A multithreaded log processor that matches events against Sigma rules. +/// A multithreaded log processor that matches events against Sigma rules and +/// evaluates Sigma correlation rules over streams of detections. /// /// The processor uses message passing to ingest log events and output detections. /// It dispatches events to matchers based on log source matching. pub struct LogProcessor { - /// Compiled matchers for all rules + /// Compiled matchers for all detection rules matchers: Vec>, + /// Correlation rules to evaluate + correlation_rules: Vec>, /// Configuration config: ProcessorConfig, } impl LogProcessor { - /// Create a new log processor with the given rules. + /// Create a new log processor with the given detection rules. + /// + /// No correlation rules are loaded; use [`LogProcessor::from_collection`] or + /// [`LogProcessor::with_config`] together with [`LogProcessor::with_correlation_rules`] + /// when correlation rules are needed. /// /// # Errors /// Returns an error if any rule cannot be compiled into a matcher. @@ -326,24 +422,85 @@ impl LogProcessor { matchers.push(Arc::new(SigmaRuleMatcher::new(rule)?)); } - Ok(Self { matchers, config }) + Ok(Self { + matchers, + correlation_rules: Vec::new(), + config, + }) + } + + /// Create a log processor from a [`crate::types::SigmaCollection`], loading all + /// detection rules and correlation rules found in the collection. + /// + /// # Errors + /// Returns an error if any detection rule cannot be compiled into a matcher. + pub fn from_collection( + collection: &crate::types::SigmaCollection, + ) -> Result { + Self::from_collection_with_config(collection, ProcessorConfig::default()) + } + + /// Create a log processor from a [`crate::types::SigmaCollection`] with custom + /// configuration, loading all detection rules and correlation rules found in the + /// collection. + /// + /// # Errors + /// Returns an error if any detection rule cannot be compiled into a matcher. + pub fn from_collection_with_config( + collection: &crate::types::SigmaCollection, + config: ProcessorConfig, + ) -> Result { + let mut matchers = Vec::new(); + let mut correlation_rules = Vec::new(); + + for doc in &collection.documents { + match doc { + SigmaDocument::Rule(rule) => { + matchers.push(Arc::new(SigmaRuleMatcher::new(rule.clone())?)); + } + SigmaDocument::Correlation(corr) => { + correlation_rules.push(Arc::new(corr.clone())); + } + } + } + + Ok(Self { + matchers, + correlation_rules, + config, + }) + } + + /// Add correlation rules to an existing processor, consuming and returning it. + pub fn with_correlation_rules( + mut self, + correlation_rules: Vec, + ) -> Self { + self.correlation_rules + .extend(correlation_rules.into_iter().map(Arc::new)); + self } /// Start processing events. /// - /// Returns a tuple of (event_sender, detection_receiver) for message passing. - /// Send events to the event_sender, and receive detections from the detection_receiver. - /// When done sending events, drop the event_sender to signal workers to shut down. - pub fn start( - &self, - ) -> (Sender, Receiver) { + /// Returns a tuple of `(event_sender, detection_receiver)` for message passing. + /// + /// - Send [`LogEvent`]s to `event_sender`. + /// - Receive [`DetectionResult`]s from `detection_receiver`. Each result is + /// either a [`DetectionResult::Rule`] (a plain detection rule matched) or a + /// [`DetectionResult::Correlation`] (a correlation rule condition was satisfied). + /// - Drop `event_sender` to signal all workers to shut down. + pub fn start(&self) -> (Sender, Receiver) { let (event_tx, event_rx) = if self.config.event_buffer_size == 0 { unbounded() } else { bounded(self.config.event_buffer_size) }; - let (detection_tx, detection_rx) = if self.config.detection_buffer_size == 0 { + // Internal channel: worker threads → correlation worker + let (raw_detection_tx, raw_detection_rx) = unbounded::(); + + let (output_tx, output_rx) = if self.config.detection_buffer_size == 0 { unbounded() } else { bounded(self.config.detection_buffer_size) @@ -352,52 +509,102 @@ impl LogProcessor { // Spawn worker threads for _ in 0..self.config.num_threads { let event_rx_clone = event_rx.clone(); - let detection_tx_clone = detection_tx.clone(); + let raw_detection_tx_clone = raw_detection_tx.clone(); let matchers = self.matchers.clone(); thread::spawn(move || { - Self::worker_thread(event_rx_clone, detection_tx_clone, matchers); + Self::worker_thread(event_rx_clone, raw_detection_tx_clone, matchers); }); } - // Drop our copies so only the senders/receivers returned to caller remain + // Drop the processor's copy of the sender/receiver so they close when workers finish drop(event_rx); - drop(detection_tx); + drop(raw_detection_tx); - (event_tx, detection_rx) + // Spawn the correlation worker + let correlation_rules = self.correlation_rules.clone(); + let late_threshold = Duration::from_secs(self.config.late_event_threshold_secs); + let output_tx_clone = output_tx.clone(); + + thread::spawn(move || { + Self::correlation_worker( + raw_detection_rx, + output_tx_clone, + correlation_rules, + late_threshold, + ); + }); + + drop(output_tx); + + (event_tx, output_rx) } - /// Worker thread function that processes events. + /// Worker thread: match incoming events against compiled rule matchers and + /// forward resulting [`Detection`]s to the internal detection channel. fn worker_thread( event_rx: Receiver, detection_tx: Sender, matchers: Vec>, ) { while let Ok(event) = event_rx.recv() { - // Try each matcher that matches the log source for matcher in &matchers { - if Self::log_source_matches(&event.log_source, &matcher.rule.logsource) { - if matcher.matches(&event.data) { - let detection = Detection { - rule: matcher.rule.clone(), - event: event.clone(), - }; - // If send fails, the receiver was dropped, so we should exit - if detection_tx.send(detection).is_err() { - return; - } + if Self::log_source_matches(&event.log_source, &matcher.rule.logsource) + && matcher.matches(&event.data) + { + let detection = Detection { + rule: matcher.rule.clone(), + event: event.clone(), + }; + if detection_tx.send(detection).is_err() { + return; } } } } } + /// Correlation worker: forward plain detections to the output channel and + /// update correlation state, emitting [`CorrelationDetection`]s when a + /// correlation rule's condition is satisfied. + fn correlation_worker( + raw_detection_rx: Receiver, + output_tx: Sender, + correlation_rules: Vec>, + late_event_threshold: Duration, + ) { + let mut engine = CorrelationEngine::new(correlation_rules, late_event_threshold); + + while let Ok(detection) = raw_detection_rx.recv() { + // Process for correlation first (borrows detection) + let corr_detections = engine.process(&detection); + + // Forward the plain detection result + if output_tx + .send(DetectionResult::Rule(detection)) + .is_err() + { + return; + } + + // Forward any triggered correlation detections + for cd in corr_detections { + if output_tx.send(DetectionResult::Correlation(cd)).is_err() { + return; + } + } + } + } + /// Check if an event's log source matches a rule's log source. /// /// Matching follows the Sigma specification: /// - An empty field in the rule matches any value in the event /// - A field value in the rule must match the corresponding event field value - fn log_source_matches(event_source: &LogSource, rule_source: &LogSource) -> bool { + pub(crate) fn log_source_matches( + event_source: &LogSource, + rule_source: &LogSource, + ) -> bool { // Check category if let Some(rule_category) = &rule_source.category { match &event_source.category { @@ -438,6 +645,514 @@ impl LogProcessor { } } +// ─── Timespan parsing ──────────────────────────────────────────────────────── + +/// Parse a Sigma timespan string (e.g. `"1h"`, `"5m"`, `"30s"`, `"2d"`, `"1w"`) +/// into a [`Duration`]. +/// +/// Supported unit suffixes: `s` (seconds), `m` (minutes), `h` (hours), +/// `d` (days), `w` (weeks). Returns `None` if the string cannot be parsed. +/// +/// # Examples +/// +/// ```rust +/// use sigma_engine::processor::parse_timespan; +/// use std::time::Duration; +/// +/// assert_eq!(parse_timespan("30s"), Some(Duration::from_secs(30))); +/// assert_eq!(parse_timespan("5m"), Some(Duration::from_secs(300))); +/// assert_eq!(parse_timespan("1h"), Some(Duration::from_secs(3600))); +/// assert_eq!(parse_timespan("2d"), Some(Duration::from_secs(172800))); +/// assert_eq!(parse_timespan("1w"), Some(Duration::from_secs(604800))); +/// assert_eq!(parse_timespan("bad"), None); +/// ``` +pub fn parse_timespan(s: &str) -> Option { + let s = s.trim(); + if s.is_empty() { + return None; + } + let (num_str, unit) = s.split_at(s.len() - 1); + let count: u64 = num_str.parse().ok()?; + let secs = match unit { + "s" => count, + "m" => count * 60, + "h" => count * 3600, + "d" => count * 86_400, + "w" => count * 604_800, + _ => return None, + }; + Some(Duration::from_secs(secs)) +} + +// ─── Correlation engine ────────────────────────────────────────────────────── + +/// A single event stored in a correlation window. +struct TimedEvent { + timestamp: DateTime, + /// The name or id of the Sigma rule that produced this detection. + rule_ref: String, + event: LogEvent, +} + +/// Sliding-window state for a single (correlation rule index, group key) pair. +#[derive(Default)] +struct CorrelationWindowState { + /// Events currently inside the time window, ordered by ascending timestamp. + events: VecDeque, +} + +/// Engine that maintains per-group sliding windows for every correlation rule +/// and checks their conditions each time a new detection arrives. +struct CorrelationEngine { + rules: Vec>, + /// Pre-parsed timespan for each rule (same index as `rules`). + timespans: Vec, + /// State keyed by `(rule_index, group_key_pairs)`. + state: HashMap<(usize, Vec<(String, String)>), CorrelationWindowState>, + late_event_threshold: Duration, +} + +impl CorrelationEngine { + fn new(rules: Vec>, late_event_threshold: Duration) -> Self { + let timespans = rules + .iter() + .map(|r| { + r.correlation + .timespan + .as_deref() + .and_then(parse_timespan) + .unwrap_or(Duration::from_secs(300)) // 5 minute default when no timespan is specified + }) + .collect(); + + Self { + rules, + timespans, + state: HashMap::new(), + late_event_threshold, + } + } + + /// Process a new detection. Returns any correlation detections that are now triggered. + fn process(&mut self, detection: &Detection) -> Vec { + let rule = &detection.rule; + let event = &detection.event; + let event_time = event.timestamp; + + let mut results = Vec::new(); + + for rule_idx in 0..self.rules.len() { + let corr_rule = Arc::clone(&self.rules[rule_idx]); + + // Check whether this detection's rule is referenced by the correlation rule + let rule_ref = match rule_ref_for(rule, &corr_rule.correlation.rules) { + Some(r) => r, + None => continue, + }; + + // Compute the group key (a sorted list of (alias/field, value) pairs) + let group_key = + compute_group_key(event, &rule_ref, &corr_rule); + + let timespan = self.timespans[rule_idx]; + + // Reject late-arriving events that are older than the late-event threshold + // relative to the most recent event in the window. + let state = self + .state + .entry((rule_idx, group_key.clone())) + .or_default(); + + if let Some(newest) = state.events.back() { + let late_cutoff = newest.timestamp + - chrono::Duration::from_std(self.late_event_threshold) + .unwrap_or(chrono::Duration::zero()); + if event_time < late_cutoff { + // Too old — discard + continue; + } + } + + // Evict events that have fallen outside the sliding window + let window_cutoff = event_time + - chrono::Duration::from_std(timespan).unwrap_or(chrono::Duration::zero()); + while state + .events + .front() + .is_some_and(|e| e.timestamp < window_cutoff) + { + state.events.pop_front(); + } + + // Insert the new event (maintain ascending timestamp order) + let insert_pos = state + .events + .iter() + .rposition(|e| e.timestamp <= event_time) + .map(|p| p + 1) + .unwrap_or(0); + state.events.insert( + insert_pos, + TimedEvent { + timestamp: event_time, + rule_ref, + event: event.clone(), + }, + ); + + // Check the correlation condition + if let Some(cd) = + check_correlation_condition(&corr_rule, group_key, &state.events) + { + results.push(cd); + } + } + + results + } +} + +/// Determine whether a fired rule is referenced by a correlation rule. +/// +/// Returns the canonical reference string (name or id) on a match, or `None`. +fn rule_ref_for(rule: &SigmaRule, corr_refs: &[String]) -> Option { + for r in corr_refs { + if let Some(name) = &rule.name + && name.eq_ignore_ascii_case(r) + { + return Some(r.clone()); + } + if let Some(id) = &rule.id + && id.eq_ignore_ascii_case(r) + { + return Some(r.clone()); + } + } + None +} + +/// Build the sorted group-key vector for a detection, resolving field aliases +/// defined in the correlation rule. +fn compute_group_key( + event: &LogEvent, + rule_ref: &str, + corr_rule: &SigmaCorrelationRule, +) -> Vec<(String, String)> { + corr_rule + .correlation + .group_by + .iter() + .map(|field| { + // Resolve alias: if there is an alias entry for `field` that maps this + // rule_ref to an actual field name, use that instead. + let resolved_field = corr_rule + .correlation + .aliases + .get(field) + .and_then(|rule_map| rule_map.get(rule_ref)) + .map(|s| s.as_str()) + .unwrap_or(field.as_str()); + + let value = event + .data + .get(resolved_field) + .cloned() + .unwrap_or_default(); + + (field.clone(), value) + }) + .collect() +} + +/// Evaluate the correlation rule's condition against the current window state. +/// +/// Returns a [`CorrelationDetection`] when the condition is satisfied, or `None` +/// when it is not. +fn check_correlation_condition( + corr_rule: &Arc, + group_key: Vec<(String, String)>, + events: &VecDeque, +) -> Option { + let group_map: HashMap = group_key.into_iter().collect(); + + let satisfied = match corr_rule.correlation.correlation_type { + CorrelationType::EventCount => { + let count = events.len() as i64; + corr_rule + .correlation + .condition + .as_ref() + .map(|c| eval_simple_condition(c, count)) + .unwrap_or(false) + } + + CorrelationType::ValueCount => { + let field = match &corr_rule.correlation.condition { + Some(CorrelationCondition::Simple(sc)) => sc.field.as_deref(), + _ => None, + }; + let distinct: HashSet<&str> = events + .iter() + .filter_map(|te| { + field.and_then(|f| te.event.data.get(f).map(|s| s.as_str())) + }) + .collect(); + let count = distinct.len() as i64; + corr_rule + .correlation + .condition + .as_ref() + .map(|c| eval_simple_condition(c, count)) + .unwrap_or(false) + } + + CorrelationType::ValueSum => { + let field = match &corr_rule.correlation.condition { + Some(CorrelationCondition::Simple(sc)) => sc.field.as_deref(), + _ => None, + }; + let sum: f64 = events + .iter() + .filter_map(|te| { + field.and_then(|f| { + te.event.data.get(f).and_then(|v| v.parse::().ok()) + }) + }) + .sum(); + corr_rule + .correlation + .condition + .as_ref() + .map(|c| eval_simple_condition(c, sum as i64)) + .unwrap_or(false) + } + + CorrelationType::ValueAvg => { + let field = match &corr_rule.correlation.condition { + Some(CorrelationCondition::Simple(sc)) => sc.field.as_deref(), + _ => None, + }; + let values: Vec = events + .iter() + .filter_map(|te| { + field.and_then(|f| { + te.event.data.get(f).and_then(|v| v.parse::().ok()) + }) + }) + .collect(); + if values.is_empty() { + false + } else { + let avg = values.iter().sum::() / values.len() as f64; + corr_rule + .correlation + .condition + .as_ref() + .map(|c| eval_simple_condition(c, avg as i64)) + .unwrap_or(false) + } + } + + CorrelationType::ValuePercentile => { + let field = match &corr_rule.correlation.condition { + Some(CorrelationCondition::Simple(sc)) => sc.field.as_deref(), + _ => None, + }; + let mut values: Vec = events + .iter() + .filter_map(|te| { + field.and_then(|f| { + te.event.data.get(f).and_then(|v| v.parse::().ok()) + }) + }) + .collect(); + if values.is_empty() { + false + } else { + // Compute the 95th percentile of field values. + // NOTE: The Sigma spec's value_percentile correlation type requires a + // percentile rank in the condition, but the current SimpleCondition type + // does not carry one. Until the type system is extended to support an + // explicit rank, the engine uses the 95th percentile (P95) as the + // default, which is the most common operational threshold. + values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); + let idx = ((0.95 * values.len() as f64).ceil() as usize).saturating_sub(1); + let percentile_val = values[idx.min(values.len() - 1)]; + corr_rule + .correlation + .condition + .as_ref() + .map(|c| eval_simple_condition(c, percentile_val as i64)) + .unwrap_or(false) + } + } + + CorrelationType::Temporal => { + let required_refs: HashSet<&str> = corr_rule + .correlation + .rules + .iter() + .map(|s| s.as_str()) + .collect(); + + let fired: HashSet<&str> = events + .iter() + .map(|te| te.rule_ref.as_str()) + .collect(); + + let all_fired = required_refs.iter().all(|r| fired.contains(r)); + + if !all_fired { + false + } else { + // Evaluate optional extended boolean condition + match &corr_rule.correlation.condition { + None => true, + Some(CorrelationCondition::Extended(expr)) => { + eval_temporal_condition(expr, &fired) + } + Some(CorrelationCondition::Simple(_)) => true, + } + } + } + + CorrelationType::TemporalOrdered => { + check_temporal_ordered(&corr_rule.correlation.rules, events) + } + }; + + if satisfied { + Some(CorrelationDetection { + rule: Arc::clone(corr_rule), + group_key: group_map, + contributing_events: events.iter().map(|te| te.event.clone()).collect(), + }) + } else { + None + } +} + +/// Evaluate a [`CorrelationCondition`] with a numeric value. +fn eval_simple_condition(condition: &CorrelationCondition, value: i64) -> bool { + let sc = match condition { + CorrelationCondition::Simple(sc) => sc, + CorrelationCondition::Extended(_) => return false, + }; + if sc.gt.is_some_and(|gt| value <= gt) { + return false; + } + if sc.gte.is_some_and(|gte| value < gte) { + return false; + } + if sc.lt.is_some_and(|lt| value >= lt) { + return false; + } + if sc.lte.is_some_and(|lte| value > lte) { + return false; + } + if sc.eq.is_some_and(|eq| value != eq) { + return false; + } + if sc.neq.is_some_and(|neq| value == neq) { + return false; + } + true +} + +/// Evaluate a temporal extended boolean condition expression against the set of +/// rule references that have fired within the current window. +fn eval_temporal_condition( + expr: &crate::types::ConditionExpression, + fired: &HashSet<&str>, +) -> bool { + use crate::types::ConditionExpression; + match expr { + ConditionExpression::Identifier(name) => fired.contains(name.as_str()), + ConditionExpression::And(l, r) => { + eval_temporal_condition(l, fired) && eval_temporal_condition(r, fired) + } + ConditionExpression::Or(l, r) => { + eval_temporal_condition(l, fired) || eval_temporal_condition(r, fired) + } + ConditionExpression::Not(e) => !eval_temporal_condition(e, fired), + ConditionExpression::OneOfThem => !fired.is_empty(), + // `all of them` in a temporal extended condition is reached only after the + // outer `all_fired` pre-check already confirmed every required rule has fired, + // so this is always satisfied here. + ConditionExpression::AllOfThem => true, + ConditionExpression::OneOfPattern(pat) => { + let pat_lower = pat.to_lowercase(); + fired.iter().any(|r| glob_matches(&pat_lower, r)) + } + ConditionExpression::AllOfPattern(pat) => { + let pat_lower = pat.to_lowercase(); + fired.iter().all(|r| glob_matches(&pat_lower, r)) + } + } +} + +/// Check whether a `temporal_ordered` correlation rule's conditions are met. +/// +/// All rules in `ordered_refs` must have fired in order: the earliest event +/// from `ordered_refs[i+1]` must not predate the earliest event from +/// `ordered_refs[i]`. +fn check_temporal_ordered( + ordered_refs: &[String], + events: &VecDeque, +) -> bool { + if ordered_refs.is_empty() { + return true; + } + + // Collect the earliest timestamp for each required rule reference + let mut first_times: Vec>> = vec![None; ordered_refs.len()]; + + for te in events { + for (i, r) in ordered_refs.iter().enumerate() { + if te.rule_ref.eq_ignore_ascii_case(r) { + let entry = &mut first_times[i]; + if entry.is_none_or(|t| te.timestamp < t) { + *entry = Some(te.timestamp); + } + } + } + } + + // All rules must have fired + let times: Vec> = match first_times.into_iter().collect::>>() { + Some(v) => v, + None => return false, + }; + + // Each rule must have fired no earlier than the rule before it + times.windows(2).all(|w| w[0] <= w[1]) +} + +/// Simple glob matching: `*` matches any sequence of characters. +fn glob_matches(pattern: &str, text: &str) -> bool { + let text = text.to_lowercase(); + if pattern == "*" { + return true; + } + // Split on `*` and match each part in order + let mut parts = pattern.split('*'); + let first = parts.next().unwrap_or(""); + if !text.starts_with(first) { + return false; + } + let mut pos = first.len(); + for part in parts { + if part.is_empty() { + continue; + } + match text[pos..].find(part) { + Some(idx) => pos += idx + part.len(), + None => return false, + } + } + true +} + #[cfg(test)] mod tests { use super::*; @@ -575,7 +1290,11 @@ detection: drop(event_tx); // Signal completion // Receive detection - let detection = detection_rx.recv().unwrap(); + let result = detection_rx.recv().unwrap(); + let detection = match result { + DetectionResult::Rule(d) => d, + _ => panic!("Expected rule detection"), + }; assert_eq!(detection.rule.title, "Test Rule"); assert_eq!(detection.event.data.get("EventID"), Some(&"4688".to_string())); } @@ -734,6 +1453,7 @@ detection: num_threads: 1, event_buffer_size: 0, detection_buffer_size: 0, + late_event_threshold_secs: 3600, }; let processor = LogProcessor::with_config(vec![rule], config).unwrap(); @@ -750,7 +1470,749 @@ detection: event_tx.send(event).unwrap(); drop(event_tx); - let detection = detection_rx.recv().unwrap(); + let result = detection_rx.recv().unwrap(); + let detection = match result { + DetectionResult::Rule(d) => d, + _ => panic!("Expected rule detection"), + }; assert_eq!(detection.rule.title, "Test Rule"); } + + // ─── Timestamp tests ────────────────────────────────────────────────────── + + #[test] + fn test_log_event_timestamp_default() { + let before = Utc::now(); + let event = LogEvent::from_fields( + LogSource { category: None, product: None, service: None }, + HashMap::new(), + ); + let after = Utc::now(); + assert!(event.get_time() >= before); + assert!(event.get_time() <= after); + } + + #[test] + fn test_log_event_with_timestamp() { + use chrono::TimeZone; + let ts = Utc.with_ymd_and_hms(2024, 6, 1, 10, 0, 0).unwrap(); + let event = LogEvent::from_fields( + LogSource { category: None, product: None, service: None }, + HashMap::new(), + ) + .with_timestamp(ts); + assert_eq!(event.get_time(), ts); + } + + #[test] + fn test_log_event_from_json_has_timestamp() { + let ls = LogSource { category: None, product: None, service: None }; + let event = LogEvent::from_json(ls, "{}").unwrap(); + // timestamp should be close to now + let diff = (Utc::now() - event.get_time()).num_seconds().abs(); + assert!(diff < 5); + } + + #[test] + fn test_log_event_from_plain_has_timestamp() { + let ls = LogSource { category: None, product: None, service: None }; + let event = LogEvent::from_plain(ls, "hello".to_string()); + let diff = (Utc::now() - event.get_time()).num_seconds().abs(); + assert!(diff < 5); + } + + #[test] + fn test_log_event_from_field_value_has_timestamp() { + let ls = LogSource { category: None, product: None, service: None }; + let event = LogEvent::from_field_value_format(ls, r#"K="V""#); + let diff = (Utc::now() - event.get_time()).num_seconds().abs(); + assert!(diff < 5); + } + + // ─── parse_timespan tests ───────────────────────────────────────────────── + + #[test] + fn test_parse_timespan_seconds() { + assert_eq!(parse_timespan("30s"), Some(Duration::from_secs(30))); + assert_eq!(parse_timespan("1s"), Some(Duration::from_secs(1))); + } + + #[test] + fn test_parse_timespan_minutes() { + assert_eq!(parse_timespan("5m"), Some(Duration::from_secs(300))); + } + + #[test] + fn test_parse_timespan_hours() { + assert_eq!(parse_timespan("1h"), Some(Duration::from_secs(3600))); + assert_eq!(parse_timespan("2h"), Some(Duration::from_secs(7200))); + } + + #[test] + fn test_parse_timespan_days() { + assert_eq!(parse_timespan("2d"), Some(Duration::from_secs(172_800))); + } + + #[test] + fn test_parse_timespan_weeks() { + assert_eq!(parse_timespan("1w"), Some(Duration::from_secs(604_800))); + } + + #[test] + fn test_parse_timespan_invalid() { + assert_eq!(parse_timespan(""), None); + assert_eq!(parse_timespan("bad"), None); + assert_eq!(parse_timespan("5x"), None); + assert_eq!(parse_timespan("abc"), None); + } + + // ─── event_count correlation tests ──────────────────────────────────────── + + fn make_detection_rule_yaml(name: &str, title: &str) -> String { + format!( + r#" +title: {title} +name: {name} +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"# + ) + } + + fn make_corr_yaml( + corr_type: &str, + rule_refs: &[&str], + group_by: &[&str], + timespan: &str, + condition: &str, + ) -> String { + let refs = rule_refs + .iter() + .map(|r| format!(" - {}", r)) + .collect::>() + .join("\n"); + let groups = if group_by.is_empty() { + " []".to_string() + } else { + group_by + .iter() + .map(|g| format!(" - {}", g)) + .collect::>() + .join("\n") + }; + format!( + r#" +title: Correlation Rule +type: correlation +correlation: + type: {corr_type} + rules: +{refs} + group-by: +{groups} + timespan: {timespan} + condition: + {condition} +"# + ) + } + + fn make_test_event(event_id: &str) -> LogEvent { + let mut data = HashMap::new(); + data.insert("EventID".to_string(), event_id.to_string()); + LogEvent::from_fields( + LogSource { + category: None, + product: Some("test".to_string()), + service: None, + }, + data, + ) + } + + #[test] + fn test_event_count_correlation_fires() { + let det_yaml = make_detection_rule_yaml("login_fail", "Login Failure"); + let corr_yaml = make_corr_yaml( + "event_count", + &["login_fail"], + &[], + "5m", + "gte: 3", + ); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Send 3 matching events — correlation should fire on the 3rd + for _ in 0..3 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let mut rule_hits = 0u32; + let mut corr_hits = 0u32; + while let Ok(result) = detection_rx.recv() { + match result { + DetectionResult::Rule(_) => rule_hits += 1, + DetectionResult::Correlation(cd) => { + corr_hits += 1; + assert_eq!(cd.rule.title, "Correlation Rule"); + assert!(!cd.contributing_events.is_empty()); + } + } + } + assert_eq!(rule_hits, 3); + // Correlation fires at count==3, then again for every subsequent event; + // since we sent exactly 3 events it fires exactly once. + assert_eq!(corr_hits, 1); + } + + #[test] + fn test_event_count_correlation_does_not_fire_below_threshold() { + let det_yaml = make_detection_rule_yaml("login_fail", "Login Failure"); + let corr_yaml = make_corr_yaml( + "event_count", + &["login_fail"], + &[], + "5m", + "gte: 5", + ); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Only 2 events — below the threshold of 5 + for _ in 0..2 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: Vec<_> = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .collect(); + assert!(corr_hits.is_empty()); + } + + // ─── value_count correlation tests ──────────────────────────────────────── + + #[test] + fn test_value_count_correlation_fires() { + // Build a custom YAML with a field in the correlation condition + let det_yaml = r#" +title: DNS Query +name: dns_query +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: Many Unique DNS Destinations +type: correlation +correlation: + type: value_count + rules: + - dns_query + group-by: [] + timespan: 5m + condition: + field: Destination + gte: 3 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // 4 events with 4 distinct destinations + for dest in &["a.com", "b.com", "c.com", "d.com"] { + let mut data = HashMap::new(); + data.insert("EventID".to_string(), "1".to_string()); + data.insert("Destination".to_string(), dest.to_string()); + let event = LogEvent::from_fields( + LogSource { category: None, product: Some("test".to_string()), service: None }, + data, + ); + event_tx.send(event).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + // Should fire at 3 distinct values (count==3) and again at 4 (count==4) + assert!(corr_hits >= 1); + } + + // ─── temporal correlation tests ─────────────────────────────────────────── + + #[test] + fn test_temporal_correlation_fires_when_all_rules_match() { + let yaml_a = r#" +title: Rule A +name: rule_a +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let yaml_b = r#" +title: Rule B +name: rule_b +logsource: + product: test +detection: + sel: + EventID: 2 + condition: sel +"#; + let corr_yaml = r#" +title: Temporal Corr +type: correlation +correlation: + type: temporal + rules: + - rule_a + - rule_b + group-by: [] + timespan: 5m +"#; + let combined = format!("{}\n---\n{}\n---\n{}", yaml_a, yaml_b, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + event_tx.send(make_test_event("1")).unwrap(); + event_tx.send(make_test_event("2")).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + #[test] + fn test_temporal_correlation_does_not_fire_with_only_one_rule() { + let yaml_a = r#" +title: Rule A +name: rule_a +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: Temporal Corr +type: correlation +correlation: + type: temporal + rules: + - rule_a + - rule_b + group-by: [] + timespan: 5m +"#; + let combined = format!("{}\n---\n{}", yaml_a, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Only rule_a fires — rule_b never does + event_tx.send(make_test_event("1")).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + // ─── temporal_ordered correlation tests ─────────────────────────────────── + + #[test] + fn test_temporal_ordered_fires_in_correct_order() { + use chrono::TimeZone; + + let yaml_a = r#" +title: Rule A +name: rule_a +logsource: + product: test +detection: + sel: + EventID: 10 + condition: sel +"#; + let yaml_b = r#" +title: Rule B +name: rule_b +logsource: + product: test +detection: + sel: + EventID: 20 + condition: sel +"#; + let corr_yaml = r#" +title: Ordered Temporal +type: correlation +correlation: + type: temporal_ordered + rules: + - rule_a + - rule_b + group-by: [] + timespan: 10m +"#; + let combined = format!("{}\n---\n{}\n---\n{}", yaml_a, yaml_b, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // a fires at t=0, b fires at t=1 — correct order + let t0 = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); + let t1 = Utc.with_ymd_and_hms(2024, 1, 1, 0, 1, 0).unwrap(); + + let mut e_a = make_test_event("10"); + e_a.timestamp = t0; + let mut e_b = make_test_event("20"); + e_b.timestamp = t1; + + event_tx.send(e_a).unwrap(); + event_tx.send(e_b).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + #[test] + fn test_temporal_ordered_does_not_fire_out_of_order() { + use chrono::TimeZone; + + let yaml_a = r#" +title: Rule A +name: rule_a +logsource: + product: test +detection: + sel: + EventID: 10 + condition: sel +"#; + let yaml_b = r#" +title: Rule B +name: rule_b +logsource: + product: test +detection: + sel: + EventID: 20 + condition: sel +"#; + let corr_yaml = r#" +title: Ordered Temporal +type: correlation +correlation: + type: temporal_ordered + rules: + - rule_a + - rule_b + group-by: [] + timespan: 10m +"#; + let combined = format!("{}\n---\n{}\n---\n{}", yaml_a, yaml_b, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // b fires BEFORE a — wrong order + let t0 = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); + let t1 = Utc.with_ymd_and_hms(2024, 1, 1, 0, 1, 0).unwrap(); + + let mut e_b = make_test_event("20"); + e_b.timestamp = t0; + let mut e_a = make_test_event("10"); + e_a.timestamp = t1; + + event_tx.send(e_b).unwrap(); + event_tx.send(e_a).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + // ─── value_sum correlation tests ────────────────────────────────────────── + + #[test] + fn test_value_sum_correlation_fires() { + let det_yaml = r#" +title: Transfer Event +name: transfer_event +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: Large Transfer Sum +type: correlation +correlation: + type: value_sum + rules: + - transfer_event + group-by: [] + timespan: 5m + condition: + field: Bytes + gte: 1000 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Sum = 400 + 700 = 1100 ≥ 1000 — should fire on 2nd event + for bytes in &["400", "700"] { + let mut data = HashMap::new(); + data.insert("EventID".to_string(), "1".to_string()); + data.insert("Bytes".to_string(), bytes.to_string()); + let event = LogEvent::from_fields( + LogSource { category: None, product: Some("test".to_string()), service: None }, + data, + ); + event_tx.send(event).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + // ─── value_avg correlation tests ────────────────────────────────────────── + + #[test] + fn test_value_avg_correlation_fires() { + let det_yaml = r#" +title: CPU Event +name: cpu_event +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: High Avg CPU +type: correlation +correlation: + type: value_avg + rules: + - cpu_event + group-by: [] + timespan: 5m + condition: + field: CPU + gte: 80 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Average = (85 + 90) / 2 = 87.5 ≥ 80 — fires on second event + for cpu in &["85", "90"] { + let mut data = HashMap::new(); + data.insert("EventID".to_string(), "1".to_string()); + data.insert("CPU".to_string(), cpu.to_string()); + let event = LogEvent::from_fields( + LogSource { category: None, product: Some("test".to_string()), service: None }, + data, + ); + event_tx.send(event).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + // ─── group_by tests ─────────────────────────────────────────────────────── + + #[test] + fn test_event_count_with_group_by_isolates_groups() { + let det_yaml = r#" +title: Login Event +name: login_event +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: Brute Force by User +type: correlation +correlation: + type: event_count + rules: + - login_event + group-by: + - User + timespan: 5m + condition: + gte: 3 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // user_alice gets 3 events → should fire + // user_bob gets 2 events → should NOT fire + let users = ["alice", "alice", "alice", "bob", "bob"]; + for user in &users { + let mut data = HashMap::new(); + data.insert("EventID".to_string(), "1".to_string()); + data.insert("User".to_string(), user.to_string()); + let event = LogEvent::from_fields( + LogSource { category: None, product: Some("test".to_string()), service: None }, + data, + ); + event_tx.send(event).unwrap(); + } + drop(event_tx); + + let corr_detections: Vec<_> = detection_rx + .iter() + .filter_map(|r| match r { + DetectionResult::Correlation(cd) => Some(cd), + _ => None, + }) + .collect(); + + // All correlation detections must be for alice + for cd in &corr_detections { + assert_eq!(cd.group_key.get("User"), Some(&"alice".to_string())); + } + // At least one correlation detection + assert!(!corr_detections.is_empty()); + } + + // ─── from_collection tests ──────────────────────────────────────────────── + + #[test] + fn test_from_collection_loads_both_rule_types() { + let det_yaml = make_detection_rule_yaml("my_rule", "My Rule"); + let corr_yaml = r#" +title: Count Corr +type: correlation +correlation: + type: event_count + rules: + - my_rule + group-by: [] + timespan: 1h + condition: + gte: 10 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + + // The processor should have both a matcher and a correlation rule loaded + assert_eq!(processor.matchers.len(), 1); + assert_eq!(processor.correlation_rules.len(), 1); + } + + // ─── late-event threshold tests ─────────────────────────────────────────── + + #[test] + fn test_late_event_is_discarded() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + let corr_yaml = r#" +title: Count +type: correlation +correlation: + type: event_count + rules: + - ev + group-by: [] + timespan: 5m + condition: + gte: 3 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let config = ProcessorConfig { + num_threads: 1, + event_buffer_size: 100, + detection_buffer_size: 100, + // Only allow events up to 60 seconds old relative to latest + late_event_threshold_secs: 60, + }; + let processor = + LogProcessor::from_collection_with_config(&collection, config).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Send two recent events (now) and one very old event (2 hours ago). + // The two recent events alone do not reach the threshold of 3. + // The old event is beyond the 60-second threshold and must be discarded. + let now = Utc::now(); + let old_ts = now - chrono::Duration::hours(2); + + let mut e1 = make_test_event("1"); + e1.timestamp = now; + let mut e2 = make_test_event("1"); + e2.timestamp = now; + let mut old_event = make_test_event("1"); + old_event.timestamp = old_ts; + + event_tx.send(e1).unwrap(); + event_tx.send(e2).unwrap(); + event_tx.send(old_event).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + // The old event must have been discarded, so the threshold of 3 is never reached. + assert_eq!(corr_hits, 0); + } } From 3547c59a5ae464c03a2ba988c89671d6dd75408a Mon Sep 17 00:00:00 2001 From: Thomas Patzke Date: Mon, 30 Mar 2026 00:04:40 +0200 Subject: [PATCH 3/4] Refactor RuleCondition to unify log source matching under a single type --- src/pipeline.rs | 58 +++++++++++++++++++++++++++++-------------------- 1 file changed, 35 insertions(+), 23 deletions(-) diff --git a/src/pipeline.rs b/src/pipeline.rs index 571accd..dac50bf 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -24,7 +24,7 @@ //! conditions: //! index: "windows" //! rule_conditions: -//! - type: logsource_product +//! - type: logsource //! product: windows //! "#; //! @@ -873,30 +873,42 @@ pub struct TransformationConfig { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "type")] pub enum RuleCondition { - /// Match rules by log source category. - #[serde(rename = "logsource_category")] - LogSourceCategory { category: String }, - - /// Match rules by log source product. - #[serde(rename = "logsource_product")] - LogSourceProduct { product: String }, - - /// Match rules by log source service. - #[serde(rename = "logsource_service")] - LogSourceService { service: String }, + /// Match rules by log source attributes. + /// + /// Any combination of `category`, `product`, and `service` can be specified. + /// Only the fields that are set are compared; unspecified fields are ignored. + /// This mirrors pySigma's `LogsourceCondition`. + #[serde(rename = "logsource")] + LogSource { + #[serde(default, skip_serializing_if = "Option::is_none")] + category: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + product: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + service: Option, + }, } impl RuleCondition { fn matches(&self, rule: &SigmaRule) -> bool { match self { - Self::LogSourceCategory { category } => { - rule.logsource.category.as_deref() == Some(category) - } - Self::LogSourceProduct { product } => { - rule.logsource.product.as_deref() == Some(product) - } - Self::LogSourceService { service } => { - rule.logsource.service.as_deref() == Some(service) + Self::LogSource { category, product, service } => { + if let Some(cat) = category + && rule.logsource.category.as_deref() != Some(cat) + { + return false; + } + if let Some(prod) = product + && rule.logsource.product.as_deref() != Some(prod) + { + return false; + } + if let Some(svc) = service + && rule.logsource.service.as_deref() != Some(svc) + { + return false; + } + true } } } @@ -1134,7 +1146,7 @@ transformations: type: field_name_prefix prefix: "proc." rule_conditions: - - type: logsource_category + - type: logsource category: process_creation "#; @@ -1406,7 +1418,7 @@ transformations: type: field_name_prefix prefix: "proc." rule_conditions: - - type: logsource_product + - type: logsource product: linux "#; @@ -2423,7 +2435,7 @@ transformations: type: field_name_prefix prefix: "s." rule_conditions: - - type: logsource_service + - type: logsource service: sysmon "#; let pipeline = ProcessingPipeline::from_yaml(yaml).unwrap(); From a84622fc6020c618283f4d5388f87bf2dd98ecbd Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 14 Jun 2026 22:28:58 +0000 Subject: [PATCH 4/4] Add tests to bring coverage from 92.41% to 95.24% (above the 95% threshold) --- src/matcher.rs | 251 +++++++++++++++++++ src/parser.rs | 93 +++++++ src/pipeline.rs | 67 ++++- src/processor.rs | 619 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 1029 insertions(+), 1 deletion(-) diff --git a/src/matcher.rs b/src/matcher.rs index 51b4218..56d5fa3 100644 --- a/src/matcher.rs +++ b/src/matcher.rs @@ -2619,5 +2619,256 @@ detection: event.insert("Count".to_string(), "42".to_string()); assert!(matcher.matches(&event)); } + + // ─── BTreeMap event matching ───────────────────────────────────────────── + + #[test] + fn test_matcher_with_btreemap_event() { + let yaml = r#" +title: Test Rule +logsource: + product: windows +detection: + selection: + EventID: 4688 + Image|contains: 'cmd' + condition: selection +"#; + let collection = SigmaCollection::from_yaml(yaml).unwrap(); + let rule = match &collection.documents[0] { + SigmaDocument::Rule(r) => r.clone(), + _ => panic!("Expected rule"), + }; + let matcher = SigmaRuleMatcher::new(rule).unwrap(); + + let mut event = std::collections::BTreeMap::new(); + event.insert("EventID".to_string(), "4688".to_string()); + event.insert("Image".to_string(), "C:\\Windows\\System32\\cmd.exe".to_string()); + + assert!(matcher.matches(&event)); + + let mut no_match = std::collections::BTreeMap::new(); + no_match.insert("EventID".to_string(), "9999".to_string()); + assert!(!matcher.matches(&no_match)); + } + + #[test] + fn test_btreemap_keyword_search() { + let yaml = r#" +title: Keyword Rule +logsource: + product: test +detection: + keywords: + - '*cmd.exe*' + condition: keywords +"#; + let collection = SigmaCollection::from_yaml(yaml).unwrap(); + let rule = match &collection.documents[0] { + SigmaDocument::Rule(r) => r.clone(), + _ => panic!("Expected rule"), + }; + let matcher = SigmaRuleMatcher::new(rule).unwrap(); + + let mut event: std::collections::BTreeMap = std::collections::BTreeMap::new(); + event.insert("CommandLine".to_string(), "C:\\Windows\\cmd.exe".to_string()); + + assert!(matcher.matches(&event)); + } + + // ─── Value::as_str and Display for non-String variants ────────────────── + + #[test] + fn test_value_as_str_variants() { + use chrono::TimeZone; + use std::net::IpAddr; + use std::str::FromStr; + + let int_val = Value::Integer(42i64); + assert_eq!(int_val.as_str(), "42"); + + let float_val = Value::Float(3.14f64); + assert!(float_val.as_str().starts_with("3.14")); + + let bool_val = Value::Boolean(true); + assert_eq!(bool_val.as_str(), "true"); + + let time_val = Value::Time(Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap()); + assert!(time_val.as_str().contains("2024")); + + let ip_val = Value::Ip(IpAddr::from_str("192.168.1.1").unwrap()); + assert_eq!(ip_val.as_str(), "192.168.1.1"); + } + + #[test] + fn test_value_display_variants() { + use chrono::TimeZone; + use std::net::IpAddr; + use std::str::FromStr; + + assert_eq!(format!("{}", Value::Integer(99i64)), "99"); + assert!(format!("{}", Value::Float(1.5f64)).starts_with("1.5")); + assert_eq!(format!("{}", Value::Boolean(false)), "false"); + assert!(format!("{}", Value::Time(Utc.with_ymd_and_hms(2024, 6, 1, 0, 0, 0).unwrap())).contains("2024")); + assert_eq!(format!("{}", Value::Ip(IpAddr::from_str("10.0.0.1").unwrap())), "10.0.0.1"); + } + + // ─── Bool pattern matching ─────────────────────────────────────────────── + + #[test] + fn test_matcher_bool_string_match() { + // When the event field contains "true" string, it should match a bool pattern + let yaml = r#" +title: Bool Rule +logsource: + product: test +detection: + selection: + Flag: true + condition: selection +"#; + let collection = SigmaCollection::from_yaml(yaml).unwrap(); + let rule = match &collection.documents[0] { + SigmaDocument::Rule(r) => r.clone(), + _ => panic!("Expected rule"), + }; + let matcher = SigmaRuleMatcher::new(rule).unwrap(); + + // String "true" should match boolean true pattern + let mut event = HashMap::new(); + event.insert("Flag".to_string(), "true".to_string()); + assert!(matcher.matches(&event)); + + // String "false" should not match boolean true pattern + let mut event2 = HashMap::new(); + event2.insert("Flag".to_string(), "false".to_string()); + assert!(!matcher.matches(&event2)); + + // Numeric value should not match a bool pattern + let mut event3 = HashMap::new(); + event3.insert("Flag".to_string(), "1".to_string()); + assert!(!matcher.matches(&event3)); + } + + // ─── Null pattern ──────────────────────────────────────────────────────── + + #[test] + fn test_matcher_null_pattern_non_string_no_match() { + // A null pattern with a non-string value type should not match + let yaml = r#" +title: Null Rule +logsource: + product: test +detection: + selection: + Field: null + condition: selection +"#; + let collection = SigmaCollection::from_yaml(yaml).unwrap(); + let rule = match &collection.documents[0] { + SigmaDocument::Rule(r) => r.clone(), + _ => panic!("Expected rule"), + }; + let matcher = SigmaRuleMatcher::new(rule).unwrap(); + + // Empty string matches null pattern + let mut event_empty = HashMap::new(); + event_empty.insert("Field".to_string(), "".to_string()); + assert!(matcher.matches(&event_empty)); + + // "null" string matches null pattern + let mut event_null = HashMap::new(); + event_null.insert("Field".to_string(), "null".to_string()); + assert!(matcher.matches(&event_null)); + + // Non-null string should not match + let mut event_val = HashMap::new(); + event_val.insert("Field".to_string(), "something".to_string()); + assert!(!matcher.matches(&event_val)); + } + + // ─── Numeric pattern type mismatches ───────────────────────────────────── + + #[test] + fn test_matcher_int_pattern_non_parseable_string() { + // A string that cannot be parsed as int should not match an int pattern + let yaml = r#" +title: Int Rule +logsource: + product: test +detection: + selection: + Count: 42 + condition: selection +"#; + let collection = SigmaCollection::from_yaml(yaml).unwrap(); + let rule = match &collection.documents[0] { + SigmaDocument::Rule(r) => r.clone(), + _ => panic!("Expected rule"), + }; + let matcher = SigmaRuleMatcher::new(rule).unwrap(); + + // Non-numeric string should not match + let mut event = HashMap::new(); + event.insert("Count".to_string(), "not-a-number".to_string()); + assert!(!matcher.matches(&event)); + } + + #[test] + fn test_matcher_float_pattern_non_parseable_string() { + // A string that cannot be parsed as float should not match a float pattern + let yaml = r#" +title: Float Rule +logsource: + product: test +detection: + selection: + Score: 3.14 + condition: selection +"#; + let collection = SigmaCollection::from_yaml(yaml).unwrap(); + let rule = match &collection.documents[0] { + SigmaDocument::Rule(r) => r.clone(), + _ => panic!("Expected rule"), + }; + let matcher = SigmaRuleMatcher::new(rule).unwrap(); + + // Non-numeric string should not match + let mut event = HashMap::new(); + event.insert("Score".to_string(), "not-a-float".to_string()); + assert!(!matcher.matches(&event)); + + // Correct float string should match + let mut event2 = HashMap::new(); + event2.insert("Score".to_string(), "3.14".to_string()); + assert!(matcher.matches(&event2)); + } + + // ─── Invalid regex (cached error) ──────────────────────────────────────── + + #[test] + fn test_matcher_invalid_regex_no_match() { + let yaml = r#" +title: Regex Rule +logsource: + product: test +detection: + selection: + Field|re: '[invalid' + condition: selection +"#; + let collection = SigmaCollection::from_yaml(yaml).unwrap(); + let rule = match &collection.documents[0] { + SigmaDocument::Rule(r) => r.clone(), + _ => panic!("Expected rule"), + }; + let matcher = SigmaRuleMatcher::new(rule).unwrap(); + + // Invalid regex should never match; calling twice exercises the cache path + let mut event = HashMap::new(); + event.insert("Field".to_string(), "anything".to_string()); + assert!(!matcher.matches(&event)); + assert!(!matcher.matches(&event)); + } } diff --git a/src/parser.rs b/src/parser.rs index 98928ca..118ced0 100644 --- a/src/parser.rs +++ b/src/parser.rs @@ -1314,4 +1314,97 @@ correlation: _ => panic!("Expected Correlation"), } } + + // ─── parse_aliases error cases ──────────────────────────────────────────── + + #[test] + fn test_parse_aliases_field_value_not_string() { + // The field name inside the alias mapping is not a string (it's a list) + let yaml = r#" +title: Test Correlation +type: sigma_correlation_rule +correlation: + type: temporal + rules: + - rule_a + - rule_b + group-by: [] + timespan: 5m + condition: "rule_a and rule_b" + aliases: + user_field: + rule_a: [not, a, string] +"#; + let result = crate::SigmaCollection::from_yaml(yaml); + assert!(result.is_err()); + } + + // ─── yaml_to_sigma_value unsupported type ───────────────────────────────── + + #[test] + fn test_yaml_to_sigma_value_sequence_error() { + // A sequence inside a detection field mapping is not a valid SigmaValue + let yaml = r#" +title: Test Rule +logsource: + product: test +detection: + selection: + Field: + - nested: value + condition: selection +"#; + // A sequence of mappings inside a field value detection is not valid + let result = crate::SigmaCollection::from_yaml(yaml); + // Either it parses as a MapList (valid) or returns an error for the unsupported type + // The key behaviour we're testing is that it doesn't panic + let _ = result; + } + + // ─── parse_correlation_section: not a mapping ───────────────────────────── + + #[test] + fn test_parse_correlation_section_extended_condition_temporal_ordered() { + // Extended condition IS allowed for temporal_ordered + let yaml = r#" +title: Temporal Ordered With Extended +type: sigma_correlation_rule +correlation: + type: temporal_ordered + rules: + - rule_a + - rule_b + group-by: [] + timespan: 5m + condition: "rule_a and rule_b" +"#; + let result = crate::SigmaCollection::from_yaml(yaml); + assert!(result.is_ok()); + } + + // ─── Correlation rule custom fields ─────────────────────────────────────── + + #[test] + fn test_parse_correlation_rule_custom_fields() { + let yaml = r#" +title: Test Correlation +type: sigma_correlation_rule +correlation: + type: event_count + rules: + - rule1 + group-by: [] + timespan: 5m + condition: + gte: 10 +custom_key: custom_value +"#; + let collection = crate::SigmaCollection::from_yaml(yaml).unwrap(); + match &collection.documents[0] { + crate::SigmaDocument::Correlation(corr) => { + assert!(corr.custom.contains_key("custom_key")); + } + _ => panic!("Expected Correlation"), + } + } } diff --git a/src/pipeline.rs b/src/pipeline.rs index dac50bf..8c789f3 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -2445,5 +2445,70 @@ transformations: assert_eq!(maps[0][0].field.as_deref(), Some("s.EventID")); } } - + + // ─── RuleCondition: product/service mismatch ───────────────────────────── + + #[test] + fn test_rule_condition_product_mismatch_skips_transformation() { + // The rule has product=windows, the condition requires product=linux + // → transformation should NOT be applied + let yaml = r#" +name: Test +transformations: + - id: pfx + type: field_name_prefix + prefix: "pfx." + rule_conditions: + - type: logsource + product: linux +"#; + let pipeline = ProcessingPipeline::from_yaml(yaml).unwrap(); + let mut rule = create_test_rule(); + pipeline.apply(&mut rule).unwrap(); + // Rule product is windows, condition requires linux → prefix NOT applied + if let SearchIdentifier::Map(items) = &rule.detection.search_identifiers["selection"] { + assert_eq!(items[0].field.as_deref(), Some("EventID")); + } + } + + #[test] + fn test_rule_condition_service_mismatch_skips_transformation() { + // The rule has no service, the condition requires service=sysmon + // → transformation should NOT be applied (service is None, not "sysmon") + let yaml = r#" +name: Test +transformations: + - id: pfx + type: field_name_prefix + prefix: "pfx." + rule_conditions: + - type: logsource + service: sysmon +"#; + let pipeline = ProcessingPipeline::from_yaml(yaml).unwrap(); + let mut rule = create_test_rule(); // service = None + pipeline.apply(&mut rule).unwrap(); + // Rule has no service → transformation NOT applied + if let SearchIdentifier::Map(items) = &rule.detection.search_identifiers["selection"] { + assert_eq!(items[0].field.as_deref(), Some("EventID")); + } + } + + // ─── replace_string with invalid regex ──────────────────────────────────── + + #[test] + fn test_replace_string_invalid_regex_error() { + let yaml = r#" +name: Test +transformations: + - id: bad_regex + type: replace_string + regex: "[invalid" + replacement: "x" +"#; + let pipeline = ProcessingPipeline::from_yaml(yaml).unwrap(); + let mut rule = create_test_rule(); + let result = pipeline.apply(&mut rule); + assert!(result.is_err()); + } } diff --git a/src/processor.rs b/src/processor.rs index 582d119..e53249c 100644 --- a/src/processor.rs +++ b/src/processor.rs @@ -2260,4 +2260,623 @@ correlation: // The old event must have been discarded, so the threshold of 3 is never reached. assert_eq!(corr_hits, 0); } + + // ─── with_correlation_rules builder ────────────────────────────────────── + + #[test] + fn test_with_correlation_rules_builder() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + let corr_yaml = r#" +title: Count +type: correlation +correlation: + type: event_count + rules: + - ev + group-by: [] + timespan: 5m + condition: + gte: 2 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + + // Load only detection rules, then add correlation rules via builder + let det_rules: Vec<_> = collection + .documents + .iter() + .filter_map(|d| match d { + crate::SigmaDocument::Rule(r) => Some(r.clone()), + _ => None, + }) + .collect(); + let corr_rules: Vec<_> = collection + .documents + .iter() + .filter_map(|d| match d { + crate::SigmaDocument::Correlation(c) => Some(c.clone()), + _ => None, + }) + .collect(); + + let processor = LogProcessor::new(det_rules) + .unwrap() + .with_correlation_rules(corr_rules); + + assert_eq!(processor.correlation_rules.len(), 1); + + let (event_tx, detection_rx) = processor.start(); + for _ in 0..2 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + // ─── log_source_matches product/service mismatch ────────────────────────── + + #[test] + fn test_log_source_product_mismatch() { + let rule_source = LogSource { + category: None, + product: Some("windows".to_string()), + service: None, + }; + let event_source = LogSource { + category: None, + product: Some("linux".to_string()), + service: None, + }; + assert!(!LogProcessor::log_source_matches(Some(&event_source), &rule_source)); + } + + #[test] + fn test_log_source_service_value_mismatch() { + let rule_source = LogSource { + category: None, + product: None, + service: Some("security".to_string()), + }; + let event_source = LogSource { + category: None, + product: None, + service: Some("application".to_string()), + }; + assert!(!LogProcessor::log_source_matches(Some(&event_source), &rule_source)); + } + + // ─── rule_ref_for by id ─────────────────────────────────────────────────── + + #[test] + fn test_correlation_by_rule_id() { + let det_yaml = r#" +title: Event By ID +id: aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: Count By ID +type: correlation +correlation: + type: event_count + rules: + - aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee + group-by: [] + timespan: 5m + condition: + gte: 2 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + for _ in 0..2 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + // ─── event window eviction ──────────────────────────────────────────────── + + #[test] + fn test_event_window_eviction() { + use chrono::TimeZone; + + let det_yaml = make_detection_rule_yaml("evict_rule", "Evict Rule"); + let corr_yaml = r#" +title: Short Window +type: correlation +correlation: + type: event_count + rules: + - evict_rule + group-by: [] + timespan: 10s + condition: + gte: 3 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Two old events (t=0) and one recent event (t=100s). + // When the recent event arrives, the old events are outside the 10s window + // and get evicted, so the count stays at 1 — below the threshold of 3. + let t0 = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); + let t1 = Utc.with_ymd_and_hms(2024, 1, 1, 0, 1, 40).unwrap(); // 100 seconds later + + let mut e1 = make_test_event("1"); + e1.timestamp = t0; + let mut e2 = make_test_event("1"); + e2.timestamp = t0; + let mut e3 = make_test_event("1"); + e3.timestamp = t1; + + event_tx.send(e1).unwrap(); + event_tx.send(e2).unwrap(); + event_tx.send(e3).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + // After eviction, window only has 1 event, below the threshold of 3 + assert_eq!(corr_hits, 0); + } + + // ─── eval_simple_condition variations ───────────────────────────────────── + + #[test] + fn test_event_count_gt_condition() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "gt: 2"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // 3 events: count=3 > 2, should fire + for _ in 0..3 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + #[test] + fn test_event_count_gt_not_satisfied() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + // "gt: 3" means count must be > 3; sending 3 events (count=3) should NOT fire + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "gt: 3"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + for _ in 0..3 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + #[test] + fn test_event_count_lt_condition() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + // "lt: 5" means count < 5; sending 2 events should fire on each + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "lt: 5"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + for _ in 0..2 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + #[test] + fn test_event_count_lt_not_satisfied() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + // "lt: 1" means count < 1; count is always >= 1, so this never fires + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "lt: 1"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + for _ in 0..3 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + #[test] + fn test_event_count_lte_not_satisfied() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + // "lte: 0" means count <= 0; count is always >= 1, so this never fires + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "lte: 0"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + for _ in 0..3 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + #[test] + fn test_event_count_eq_condition() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + // "eq: 3" means count == 3; sending exactly 3 events should fire once + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "eq: 3"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + for _ in 0..3 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + #[test] + fn test_event_count_eq_not_satisfied() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + // "eq: 5" means count == 5; sending 3 events should NOT fire + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "eq: 5"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + for _ in 0..3 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + #[test] + fn test_event_count_neq_condition() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + // "neq: 2" means count != 2; sending 3 events (count=3) should fire + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "neq: 2"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + for _ in 0..3 { + event_tx.send(make_test_event("1")).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + #[test] + fn test_event_count_neq_not_satisfied() { + let det_yaml = make_detection_rule_yaml("ev", "Event"); + // "neq: 1" means count != 1; sending exactly 1 event (count=1) should NOT fire + let corr_yaml = make_corr_yaml("event_count", &["ev"], &[], "5m", "neq: 1"); + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + event_tx.send(make_test_event("1")).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + // ─── value_percentile correlation tests ─────────────────────────────────── + + #[test] + fn test_value_percentile_correlation_fires() { + let det_yaml = r#" +title: Latency Event +name: latency_event +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: High P95 Latency +type: correlation +correlation: + type: value_percentile + rules: + - latency_event + group-by: [] + timespan: 5m + condition: + field: Latency + gte: 900 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Send 10 latency events; P95 of these should be >= 900 + for ms in &[100, 200, 300, 400, 500, 600, 700, 800, 900, 1000] { + let mut data = HashMap::new(); + data.insert("EventID".to_string(), "1".to_string()); + data.insert("Latency".to_string(), ms.to_string()); + let event = LogEvent::from_fields( + LogSource { category: None, product: Some("test".to_string()), service: None }, + data, + ); + event_tx.send(event).unwrap(); + } + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + #[test] + fn test_value_percentile_empty_field_no_fire() { + let det_yaml = r#" +title: Event +name: perc_event +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: Percentile +type: correlation +correlation: + type: value_percentile + rules: + - perc_event + group-by: [] + timespan: 5m + condition: + field: MissingField + gte: 100 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Event has no "MissingField" → values list is empty → false + event_tx.send(make_test_event("1")).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + // ─── value_avg empty field ──────────────────────────────────────────────── + + #[test] + fn test_value_avg_empty_field_no_fire() { + let det_yaml = r#" +title: Avg Event +name: avg_event +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let corr_yaml = r#" +title: Avg Empty +type: correlation +correlation: + type: value_avg + rules: + - avg_event + group-by: [] + timespan: 5m + condition: + field: MissingField + gte: 80 +"#; + let combined = format!("{}\n---\n{}", det_yaml, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // No "MissingField" → empty values → returns false + event_tx.send(make_test_event("1")).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } + + // ─── temporal with extended condition ───────────────────────────────────── + + #[test] + fn test_temporal_with_extended_condition() { + let yaml_a = r#" +title: Rule A +name: rule_a +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let yaml_b = r#" +title: Rule B +name: rule_b +logsource: + product: test +detection: + sel: + EventID: 2 + condition: sel +"#; + // Extended condition on temporal: both must fire + let corr_yaml = r#" +title: Temporal With Extended +type: correlation +correlation: + type: temporal + rules: + - rule_a + - rule_b + group-by: [] + timespan: 5m + condition: "rule_a and rule_b" +"#; + let combined = format!("{}\n---\n{}\n---\n{}", yaml_a, yaml_b, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + event_tx.send(make_test_event("1")).unwrap(); + event_tx.send(make_test_event("2")).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert!(corr_hits >= 1); + } + + #[test] + fn test_temporal_with_extended_condition_not_satisfied() { + let yaml_a = r#" +title: Rule A +name: rule_a +logsource: + product: test +detection: + sel: + EventID: 1 + condition: sel +"#; + let yaml_b = r#" +title: Rule B +name: rule_b +logsource: + product: test +detection: + sel: + EventID: 2 + condition: sel +"#; + // Extended condition requires both, but only rule_a fires + let corr_yaml = r#" +title: Temporal Extended Not Met +type: correlation +correlation: + type: temporal + rules: + - rule_a + - rule_b + group-by: [] + timespan: 5m + condition: "rule_a and rule_b" +"#; + let combined = format!("{}\n---\n{}\n---\n{}", yaml_a, yaml_b, corr_yaml); + let collection = crate::SigmaCollection::from_yaml(&combined).unwrap(); + let processor = LogProcessor::from_collection(&collection).unwrap(); + let (event_tx, detection_rx) = processor.start(); + + // Only rule_a fires — extended AND requires both + event_tx.send(make_test_event("1")).unwrap(); + drop(event_tx); + + let corr_hits: u32 = detection_rx + .iter() + .filter(|r| matches!(r, DetectionResult::Correlation(_))) + .count() as u32; + assert_eq!(corr_hits, 0); + } }