rsigma_eval/engine/mod.rs
1//! Rule evaluation engine with logsource routing.
2//!
3//! The `Engine` manages a set of compiled Sigma rules and evaluates events
4//! against them. It supports optional logsource-based pre-filtering to
5//! reduce the number of rules evaluated per event.
6
7pub(crate) mod bloom_index;
8#[cfg(feature = "daachorse-index")]
9pub(crate) mod cross_rule_ac;
10mod filters;
11#[cfg(test)]
12mod tests;
13
14use std::sync::atomic::{AtomicU64, Ordering};
15
16use rsigma_parser::{
17 ConditionExpr, FilterRule, FilterRuleTarget, LogSource, SigmaCollection, SigmaRule,
18};
19
20use crate::compiler::{CompiledRule, compile_detection, compile_rule, evaluate_rule_with_bloom};
21use crate::error::{EvalError, Result};
22use crate::event::Event;
23use crate::logsource::LogSourceExtractor;
24use crate::pipeline::{Pipeline, apply_pipelines};
25use crate::result::{EvaluationResult, MatchDetailLevel};
26use crate::rule_index::RuleIndex;
27
28use bloom_index::{BloomCache, FieldBloomIndex};
29
30use filters::{
31 filter_logsource_contains, logsource_compatible, logsource_matches,
32 rewrite_condition_identifiers,
33};
34
35/// The main rule evaluation engine.
36///
37/// Holds a set of compiled rules and provides methods to evaluate events
38/// against them. Supports optional logsource routing for performance.
39///
40/// # Example
41///
42/// ```rust
43/// use rsigma_parser::parse_sigma_yaml;
44/// use rsigma_eval::{Engine, Event};
45/// use rsigma_eval::event::JsonEvent;
46/// use serde_json::json;
47///
48/// let yaml = r#"
49/// title: Detect Whoami
50/// logsource:
51/// product: windows
52/// category: process_creation
53/// detection:
54/// selection:
55/// CommandLine|contains: 'whoami'
56/// condition: selection
57/// level: medium
58/// "#;
59///
60/// let collection = parse_sigma_yaml(yaml).unwrap();
61/// let mut engine = Engine::new();
62/// engine.add_collection(&collection).unwrap();
63///
64/// let event_val = json!({"CommandLine": "cmd /c whoami"});
65/// let event = JsonEvent::borrow(&event_val);
66/// let matches = engine.evaluate(&event);
67/// assert_eq!(matches.len(), 1);
68/// assert_eq!(matches[0].header.rule_title, "Detect Whoami");
69/// ```
70pub struct Engine {
71 rules: Vec<CompiledRule>,
72 pipelines: Vec<Pipeline>,
73 /// Global override: include the full event JSON in all match results.
74 /// When `true`, overrides per-rule `rsigma.include_event` custom attributes.
75 include_event: bool,
76 /// Verbosity of the match detail recorded on detection results.
77 /// `Off` by default, which preserves the historical `{ field, value }`
78 /// wire shape. See [`Engine::set_match_detail`].
79 match_detail: MatchDetailLevel,
80 /// Monotonic counter used to namespace injected filter detections,
81 /// preventing key collisions when multiple filters share detection names.
82 filter_counter: usize,
83 /// Inverted index mapping `(field, exact_value)` to candidate rule indices.
84 /// Rebuilt after every rule mutation (add, filter).
85 rule_index: RuleIndex,
86 /// Per-field bloom filter over positive substring needles. Rebuilt
87 /// alongside `rule_index`. Consulted only when `bloom_prefilter` is
88 /// enabled.
89 bloom_index: FieldBloomIndex,
90 /// Toggle for bloom pre-filtering. Off by default: the per-event probe
91 /// overhead exceeds the savings on rule sets where most events overlap
92 /// with at least one needle's trigrams. Workloads with many substring
93 /// rules and mostly-non-matching events (e.g. high-volume telemetry
94 /// streams against an active threat-intel ruleset) opt in via
95 /// [`Engine::set_bloom_prefilter`].
96 bloom_prefilter: bool,
97 /// Memory budget the bloom builder is allowed to consume across all
98 /// per-field filters. `None` means use the crate default
99 /// (`bloom_index::DEFAULT_MAX_TOTAL_BYTES`, 1 MB).
100 bloom_max_bytes: Option<usize>,
101 /// Opt-in event-logsource extractor for conflict-based rule pruning.
102 /// `None` (default) leaves the hot path unchanged; when `Some`, the
103 /// engine extracts each event's logsource once and skips rules whose
104 /// logsource conflicts (see [`Engine::set_logsource_extractor`]).
105 logsource_extractor: Option<LogSourceExtractor>,
106 /// Monotonic count of always-evaluated rules skipped because their
107 /// product conflicts with the event's. Incremented only when an extractor
108 /// is set; surfaced via [`Engine::logsource_pruned_total`].
109 logsource_pruned: AtomicU64,
110 /// Monotonic count of `evaluate` calls where the extractor produced no
111 /// logsource at all (fail-open: every rule was evaluated). Surfaced via
112 /// [`Engine::logsource_absent_total`].
113 logsource_absent: AtomicU64,
114 /// Cross-rule Aho-Corasick index for substring patterns, gated on the
115 /// `daachorse-index` feature. Built only when [`cross_rule_ac_enabled`]
116 /// is `true`; [`cross_rule_ac_prunable`] is the conservative per-rule
117 /// flag computed at the same time so the `evaluate` hot path can drop
118 /// rules safely.
119 ///
120 /// [`cross_rule_ac_enabled`]: Self::cross_rule_ac_enabled
121 /// [`cross_rule_ac_prunable`]: Self::cross_rule_ac_prunable
122 #[cfg(feature = "daachorse-index")]
123 cross_rule_ac_index: cross_rule_ac::CrossRuleAcIndex,
124 /// Toggle for the cross-rule AC pre-filter. Off by default; the index
125 /// only pays off on rule sets > 5K rules with many shared substring
126 /// patterns. See [`Engine::set_cross_rule_ac`].
127 #[cfg(feature = "daachorse-index")]
128 cross_rule_ac_enabled: bool,
129 /// Per-rule conservative AC-prunability flag. `true` iff the rule's
130 /// firing requires at least one positive substring match (no `Exact`,
131 /// `Regex`, `Numeric`, `Not`, etc.), so dropping the rule on a
132 /// "no AC hit" verdict is provably correct.
133 #[cfg(feature = "daachorse-index")]
134 cross_rule_ac_prunable: Vec<bool>,
135}
136
137impl Engine {
138 /// Create a new empty engine.
139 pub fn new() -> Self {
140 Engine {
141 rules: Vec::new(),
142 pipelines: Vec::new(),
143 include_event: false,
144 match_detail: MatchDetailLevel::Off,
145 filter_counter: 0,
146 rule_index: RuleIndex::empty(),
147 bloom_index: FieldBloomIndex::empty(),
148 bloom_prefilter: false,
149 bloom_max_bytes: None,
150 logsource_extractor: None,
151 logsource_pruned: AtomicU64::new(0),
152 logsource_absent: AtomicU64::new(0),
153 #[cfg(feature = "daachorse-index")]
154 cross_rule_ac_index: cross_rule_ac::CrossRuleAcIndex::empty(),
155 #[cfg(feature = "daachorse-index")]
156 cross_rule_ac_enabled: false,
157 #[cfg(feature = "daachorse-index")]
158 cross_rule_ac_prunable: Vec::new(),
159 }
160 }
161
162 /// Create a new engine with a pipeline.
163 pub fn new_with_pipeline(pipeline: Pipeline) -> Self {
164 Engine {
165 rules: Vec::new(),
166 pipelines: vec![pipeline],
167 include_event: false,
168 match_detail: MatchDetailLevel::Off,
169 filter_counter: 0,
170 rule_index: RuleIndex::empty(),
171 bloom_index: FieldBloomIndex::empty(),
172 bloom_prefilter: false,
173 bloom_max_bytes: None,
174 logsource_extractor: None,
175 logsource_pruned: AtomicU64::new(0),
176 logsource_absent: AtomicU64::new(0),
177 #[cfg(feature = "daachorse-index")]
178 cross_rule_ac_index: cross_rule_ac::CrossRuleAcIndex::empty(),
179 #[cfg(feature = "daachorse-index")]
180 cross_rule_ac_enabled: false,
181 #[cfg(feature = "daachorse-index")]
182 cross_rule_ac_prunable: Vec::new(),
183 }
184 }
185
186 /// Enable or disable bloom-filter pre-filtering of positive substring
187 /// detection items.
188 ///
189 /// When enabled, `evaluate*` short-circuits any positive substring
190 /// matcher (`Contains` / `StartsWith` / `EndsWith` / `AhoCorasickSet`,
191 /// alone or wrapped in `CaseInsensitiveGroup`) whose field cannot
192 /// possibly contain a needle trigram.
193 ///
194 /// Disabled by default. The per-event probe (trigram extraction +
195 /// double hashing) costs ~1 µs on a typical CommandLine field, which
196 /// outweighs the savings on rule sets where most events overlap with
197 /// at least one needle. Enable for workloads that pair many substring
198 /// rules with mostly-non-matching events; benchmark with
199 /// `eval_bloom_rejection` before flipping it on in production.
200 pub fn set_bloom_prefilter(&mut self, enabled: bool) {
201 self.bloom_prefilter = enabled;
202 }
203
204 /// Returns whether bloom pre-filtering is currently enabled.
205 pub fn bloom_prefilter_enabled(&self) -> bool {
206 self.bloom_prefilter
207 }
208
209 /// Set the memory budget for the per-field bloom index.
210 ///
211 /// Must be called **before** `add_collection` / `add_rule` for the new
212 /// budget to take effect on the existing rule set; otherwise it is
213 /// applied at the next index rebuild. The default budget is 1 MB,
214 /// shared across all per-field filters. Lower the cap on memory-
215 /// constrained deployments; raise it for large rule sets where the
216 /// default starts evicting useful filters.
217 pub fn set_bloom_max_bytes(&mut self, max_bytes: usize) {
218 self.bloom_max_bytes = Some(max_bytes);
219 if !self.rules.is_empty() {
220 self.rebuild_index();
221 }
222 }
223
224 /// Returns the configured bloom memory budget, if one has been set
225 /// explicitly. `None` means the crate default (1 MB) is in use.
226 pub fn bloom_max_bytes(&self) -> Option<usize> {
227 self.bloom_max_bytes
228 }
229
230 /// Enable or disable opt-in, conflict-based logsource pruning.
231 ///
232 /// When set to `Some`, `evaluate` extracts each event's logsource once via
233 /// the [`LogSourceExtractor`] and skips any candidate rule whose logsource
234 /// conflicts with the event's (a dimension set on both sides that
235 /// disagrees). A dimension unset on either side is a wildcard, so an event
236 /// tagged only `product: windows` skips `product: linux` rules while still
237 /// evaluating Windows-category and logsource-less rules.
238 ///
239 /// Disabled by default (`None`), leaving the hot path unchanged. Pruning
240 /// fails open: an event with no extractable logsource evaluates every
241 /// rule. The extractor is read on every `evaluate` call, so it can be
242 /// swapped at runtime (e.g. carried across a hot-reload).
243 pub fn set_logsource_extractor(&mut self, extractor: Option<LogSourceExtractor>) {
244 self.logsource_extractor = extractor;
245 }
246
247 /// Returns the configured logsource extractor, if any. `None` means
248 /// logsource pruning is disabled.
249 pub fn logsource_extractor(&self) -> Option<&LogSourceExtractor> {
250 self.logsource_extractor.as_ref()
251 }
252
253 /// Total always-evaluated rules skipped by logsource product pruning since
254 /// engine creation. Zero unless an extractor is set.
255 pub fn logsource_pruned_total(&self) -> u64 {
256 self.logsource_pruned.load(Ordering::Relaxed)
257 }
258
259 /// Total `evaluate` calls where the extractor produced no logsource and
260 /// pruning failed open (every rule evaluated). Zero unless an extractor
261 /// is set.
262 pub fn logsource_absent_total(&self) -> u64 {
263 self.logsource_absent.load(Ordering::Relaxed)
264 }
265
266 /// Static view of how many loaded rules are eligible (logsource-compatible)
267 /// versus pruned (conflicting) for `event_logsource`, returned as
268 /// `(eligible, pruned)`. Used to report how much of a ruleset a given
269 /// logsource (for example a schema's implied logsource) actually evaluates,
270 /// independent of any specific event's field values.
271 pub fn logsource_eligibility(&self, event_logsource: &LogSource) -> (usize, usize) {
272 let mut eligible = 0;
273 let mut pruned = 0;
274 for rule in &self.rules {
275 if logsource_compatible(&rule.logsource, event_logsource) {
276 eligible += 1;
277 } else {
278 pruned += 1;
279 }
280 }
281 (eligible, pruned)
282 }
283
284 /// Enable or disable the cross-rule Aho-Corasick pre-filter.
285 ///
286 /// When enabled, the engine builds a single per-field
287 /// `DoubleArrayAhoCorasick` automaton over every positive substring
288 /// needle from every rule and drops AC-prunable rules from the
289 /// candidate set when none of their patterns hit the event.
290 ///
291 /// Off by default. Pays off on large rule sets (> ~5K rules) with many
292 /// shared substring patterns (threat-intel feeds, IOC packs). For
293 /// smaller rule sets the per-rule [`AhoCorasickSet`] matcher already
294 /// handles the workload optimally; the cross-rule index only adds
295 /// build-time and lookup overhead. Benchmark with `eval_cross_rule_ac`
296 /// against representative rule sets before enabling in production.
297 ///
298 /// Available behind the `daachorse-index` Cargo feature.
299 ///
300 /// [`AhoCorasickSet`]: crate::matcher::CompiledMatcher::AhoCorasickSet
301 #[cfg(feature = "daachorse-index")]
302 pub fn set_cross_rule_ac(&mut self, enabled: bool) {
303 self.cross_rule_ac_enabled = enabled;
304 if enabled && !self.rules.is_empty() {
305 self.rebuild_index();
306 }
307 }
308
309 /// Returns whether the cross-rule AC pre-filter is currently enabled.
310 /// Available behind the `daachorse-index` Cargo feature.
311 #[cfg(feature = "daachorse-index")]
312 pub fn cross_rule_ac_enabled(&self) -> bool {
313 self.cross_rule_ac_enabled
314 }
315
316 /// Set global `include_event` — when `true`, all match results include
317 /// the full event JSON regardless of per-rule custom attributes.
318 pub fn set_include_event(&mut self, include: bool) {
319 self.include_event = include;
320 }
321
322 /// Set the match-detail verbosity for detection results.
323 ///
324 /// `Off` (default) records each match as `{ field, value }`, identical to
325 /// pre-enrichment releases. `Summary` adds the originating selection, the
326 /// matcher kind, and case sensitivity, and reports keyword and absence
327 /// matches that `Off` omits. `Full` additionally records the pattern that
328 /// fired. The extra work runs only when a rule matches and only above
329 /// `Off`, so the default hot path is unchanged.
330 pub fn set_match_detail(&mut self, level: MatchDetailLevel) {
331 self.match_detail = level;
332 }
333
334 /// Returns the configured match-detail verbosity.
335 pub fn match_detail(&self) -> MatchDetailLevel {
336 self.match_detail
337 }
338
339 /// Add a pipeline to the engine.
340 ///
341 /// Pipelines are applied to rules during `add_rule` / `add_collection`.
342 /// Only affects rules added **after** this call.
343 pub fn add_pipeline(&mut self, pipeline: Pipeline) {
344 self.pipelines.push(pipeline);
345 self.pipelines.sort_by_key(|p| p.priority);
346 }
347
348 /// Add a single parsed Sigma rule.
349 ///
350 /// If pipelines are set, the rule is cloned and transformed before
351 /// compilation. The rule index folds the new rule incrementally; the
352 /// bloom index also folds it incrementally and only triggers a full
353 /// rebuild when its doubling watermark is reached, so this call is
354 /// amortized O(1) per rule. With the `daachorse-index` feature
355 /// enabled **and** the cross-rule AC index turned on at runtime, the
356 /// call falls back to a full rebuild because the daachorse automaton
357 /// has no incremental update path.
358 pub fn add_rule(&mut self, rule: &SigmaRule) -> Result<()> {
359 let compiled = self.compile_with_pipelines(rule)?;
360 self.rules.push(compiled);
361 self.index_append_last_rule();
362 Ok(())
363 }
364
365 /// Add many parsed Sigma rules in a single batch.
366 ///
367 /// Each rule is compiled (with the engine's pipelines applied, if any)
368 /// and pushed onto the rule set. Compilation errors are collected and
369 /// returned as `(rule_index_in_input, error)` pairs without aborting the
370 /// batch; rules that did compile remain loaded. The inverted index and
371 /// per-field bloom filter are rebuilt **once** at the end of the batch.
372 ///
373 /// Prefer this over a loop of [`Engine::add_rule`] when loading large
374 /// rule sets: the per-call rebuild is O(N) in the total rule count, so
375 /// per-rule adds turn a 3K-rule corpus into O(N²) work.
376 pub fn add_rules<'a, I>(&mut self, rules: I) -> Vec<(usize, EvalError)>
377 where
378 I: IntoIterator<Item = &'a SigmaRule>,
379 {
380 let mut errors = Vec::new();
381 for (idx, rule) in rules.into_iter().enumerate() {
382 match self.compile_with_pipelines(rule) {
383 Ok(compiled) => self.rules.push(compiled),
384 Err(e) => errors.push((idx, e)),
385 }
386 }
387 self.rebuild_index();
388 errors
389 }
390
391 /// Add all detection rules from a parsed collection, then apply filters.
392 ///
393 /// Filter rules modify referenced detection rules by appending exclusion
394 /// conditions. Correlation rules are handled by `CorrelationEngine`.
395 /// The inverted index is rebuilt once after all rules and filters are loaded.
396 pub fn add_collection(&mut self, collection: &SigmaCollection) -> Result<()> {
397 for rule in &collection.rules {
398 let compiled = self.compile_with_pipelines(rule)?;
399 self.rules.push(compiled);
400 }
401 for filter in &collection.filters {
402 self.apply_filter_no_rebuild(filter)?;
403 }
404 self.rebuild_index();
405 Ok(())
406 }
407
408 /// Compile a rule, applying any configured pipelines first. Shared by
409 /// the single- and batched-add paths so they stay behaviourally
410 /// identical.
411 fn compile_with_pipelines(&self, rule: &SigmaRule) -> Result<CompiledRule> {
412 if self.pipelines.is_empty() {
413 compile_rule(rule)
414 } else {
415 let mut transformed = rule.clone();
416 apply_pipelines(&self.pipelines, &mut transformed)?;
417 compile_rule(&transformed)
418 }
419 }
420
421 /// Add all detection rules from a collection, applying the given pipelines.
422 ///
423 /// This is a convenience method that temporarily sets pipelines, adds the
424 /// collection, then clears them. The inverted index is rebuilt once after
425 /// all rules and filters are loaded.
426 pub fn add_collection_with_pipelines(
427 &mut self,
428 collection: &SigmaCollection,
429 pipelines: &[Pipeline],
430 ) -> Result<()> {
431 let prev = std::mem::take(&mut self.pipelines);
432 self.pipelines = pipelines.to_vec();
433 self.pipelines.sort_by_key(|p| p.priority);
434 let result = self.add_collection(collection);
435 self.pipelines = prev;
436 result
437 }
438
439 /// Apply a filter rule to all referenced detection rules and rebuild the index.
440 pub fn apply_filter(&mut self, filter: &FilterRule) -> Result<()> {
441 self.apply_filter_no_rebuild(filter)?;
442 self.rebuild_index();
443 Ok(())
444 }
445
446 /// Apply a filter rule without rebuilding the index.
447 /// Used internally when multiple mutations are batched.
448 fn apply_filter_no_rebuild(&mut self, filter: &FilterRule) -> Result<()> {
449 // Compile filter detections
450 let mut filter_detections = Vec::new();
451 for (name, detection) in &filter.detection.named {
452 let compiled = compile_detection(detection)?;
453 filter_detections.push((name.clone(), compiled));
454 }
455
456 if filter_detections.is_empty() {
457 return Ok(());
458 }
459
460 let fc = self.filter_counter;
461 self.filter_counter += 1;
462
463 // Rewrite the filter's own condition expression with namespaced identifiers
464 // so that `selection` becomes `__filter_0_selection`, etc.
465 let rewritten_cond = if let Some(cond_expr) = filter.detection.conditions.first() {
466 rewrite_condition_identifiers(cond_expr, fc)
467 } else {
468 // No explicit condition: AND all detections (legacy fallback)
469 if filter_detections.len() == 1 {
470 ConditionExpr::Identifier(format!("__filter_{fc}_{}", filter_detections[0].0))
471 } else {
472 ConditionExpr::And(
473 filter_detections
474 .iter()
475 .map(|(name, _)| ConditionExpr::Identifier(format!("__filter_{fc}_{name}")))
476 .collect(),
477 )
478 }
479 };
480
481 // Find and modify referenced rules
482 let mut matched_any = false;
483 for rule in &mut self.rules {
484 let rule_matches = match &filter.rules {
485 FilterRuleTarget::Any => true,
486 FilterRuleTarget::Specific(refs) => refs
487 .iter()
488 .any(|r| rule.id.as_deref() == Some(r.as_str()) || rule.title == *r),
489 };
490
491 // Also check logsource compatibility if the filter specifies one
492 if rule_matches {
493 if let Some(ref filter_ls) = filter.logsource
494 && !filter_logsource_contains(filter_ls, &rule.logsource)
495 {
496 continue;
497 }
498
499 // Inject filter detections into the rule
500 for (name, compiled) in &filter_detections {
501 rule.detections
502 .insert(format!("__filter_{fc}_{name}"), compiled.clone());
503 }
504
505 // Wrap each existing rule condition with the filter condition
506 rule.conditions = rule
507 .conditions
508 .iter()
509 .map(|cond| ConditionExpr::And(vec![cond.clone(), rewritten_cond.clone()]))
510 .collect();
511 matched_any = true;
512 }
513 }
514
515 if let FilterRuleTarget::Specific(_) = &filter.rules
516 && !matched_any
517 {
518 log::warn!(
519 "filter '{}' references rules {:?} but none matched any loaded rule",
520 filter.title,
521 filter.rules
522 );
523 }
524
525 Ok(())
526 }
527
528 /// Add a pre-compiled rule directly. The rule index folds the new
529 /// rule incrementally; the bloom index also folds it incrementally
530 /// and only triggers a full rebuild when its doubling watermark is
531 /// reached, so this call is amortized O(1) per rule. With the
532 /// cross-rule AC index enabled (`daachorse-index` feature, runtime
533 /// toggle), this falls back to a full rebuild because daachorse has
534 /// no incremental update path.
535 pub fn add_compiled_rule(&mut self, rule: CompiledRule) {
536 self.rules.push(rule);
537 self.index_append_last_rule();
538 }
539
540 /// Add many pre-compiled rules in a single batch. The inverted index
541 /// and bloom filter are rebuilt exactly once at the end, regardless of
542 /// how many rules are appended.
543 pub fn extend_compiled_rules<I>(&mut self, rules: I)
544 where
545 I: IntoIterator<Item = CompiledRule>,
546 {
547 self.rules.extend(rules);
548 self.rebuild_index();
549 }
550
551 /// Rebuild every per-engine index from the current rule set.
552 ///
553 /// Used by batched rule loads (`add_rules`, `extend_compiled_rules`,
554 /// `add_collection`) and by mutations that rewrite existing rules
555 /// (`apply_filter`), where rebuilding once over the final shape is
556 /// cheaper than maintaining incremental state across mutations. The
557 /// single-rule paths use [`Engine::index_append_last_rule`] instead.
558 fn rebuild_index(&mut self) {
559 self.rule_index = RuleIndex::build(&self.rules);
560 self.bloom_index = match self.bloom_max_bytes {
561 Some(budget) => FieldBloomIndex::build_with_budget(&self.rules, budget),
562 None => FieldBloomIndex::build(&self.rules),
563 };
564 #[cfg(feature = "daachorse-index")]
565 {
566 if self.cross_rule_ac_enabled {
567 self.cross_rule_ac_index = cross_rule_ac::CrossRuleAcIndex::build(&self.rules);
568 self.cross_rule_ac_prunable = self
569 .rules
570 .iter()
571 .map(cross_rule_ac::rule_is_ac_prunable)
572 .collect();
573 } else {
574 self.cross_rule_ac_index = cross_rule_ac::CrossRuleAcIndex::empty();
575 self.cross_rule_ac_prunable.clear();
576 }
577 }
578 }
579
580 /// Fold the rule most recently pushed onto `self.rules` into the
581 /// inverted and bloom indexes incrementally. Cost is bounded by the
582 /// new rule's detection tree size, not by the total rule count.
583 ///
584 /// The bloom index periodically forces a full rebuild via its
585 /// doubling watermark to re-enforce the memory budget and reset the
586 /// FPR drift that incremental inserts accumulate. Cross-rule AC
587 /// (daachorse) has no incremental story, so when it is enabled this
588 /// call falls back to [`Engine::rebuild_index`].
589 fn index_append_last_rule(&mut self) {
590 #[cfg(feature = "daachorse-index")]
591 {
592 if self.cross_rule_ac_enabled {
593 self.rebuild_index();
594 return;
595 }
596 }
597
598 let new_idx = self.rules.len() - 1;
599 let rule = &self.rules[new_idx];
600 self.rule_index.append_rule(new_idx, rule);
601 self.bloom_index.append_rule(rule);
602
603 if self.bloom_index.should_rebuild(self.rules.len()) {
604 self.bloom_index = match self.bloom_max_bytes {
605 Some(budget) => FieldBloomIndex::build_with_budget(&self.rules, budget),
606 None => FieldBloomIndex::build(&self.rules),
607 };
608 }
609 }
610
611 /// Evaluate an event against candidate rules using the inverted index.
612 ///
613 /// When a logsource extractor is configured (see
614 /// [`Engine::set_logsource_extractor`]) the event's logsource is derived
615 /// from it and used for conflict-based pruning.
616 pub fn evaluate<E: Event>(&self, event: &E) -> Vec<EvaluationResult> {
617 let event_logsource = self
618 .logsource_extractor
619 .as_ref()
620 .map(|ex| ex.extract(event));
621 self.evaluate_inner(event, event_logsource.as_ref())
622 }
623
624 /// Evaluate an event with a caller-resolved event logsource for
625 /// conflict-based pruning, bypassing the engine's own extractor.
626 ///
627 /// The schema router uses this to feed a per-event logsource resolved from
628 /// the event's explicit fields plus the recognized schema's implied
629 /// logsource, so cross-product rules are pruned even when the event carries
630 /// no explicit `product`/`service`/`category` field. Pruning is
631 /// conflict-based: a rule is skipped only when a dimension is set on both
632 /// the rule and `event_logsource` and the values differ.
633 pub fn evaluate_pruned<E: Event>(
634 &self,
635 event: &E,
636 event_logsource: &LogSource,
637 ) -> Vec<EvaluationResult> {
638 self.evaluate_inner(event, Some(event_logsource))
639 }
640
641 fn evaluate_inner<E: Event>(
642 &self,
643 event: &E,
644 event_logsource: Option<&LogSource>,
645 ) -> Vec<EvaluationResult> {
646 if self.bloom_prefilter {
647 self.evaluate_with_bloom_path(event, event_logsource)
648 } else {
649 self.evaluate_no_bloom_path(event, event_logsource)
650 }
651 }
652
653 /// Build the cross-rule AC keep-mask for `event`, or `None` when the
654 /// cross-rule index is disabled or empty (no filtering needed).
655 ///
656 /// `Some(mask)` answers "should this rule survive the cross-rule AC
657 /// filter": `mask[idx] = true` means keep, `false` means drop.
658 /// Non-AC-prunable rules are always kept.
659 #[cfg(feature = "daachorse-index")]
660 fn cross_rule_ac_keep_mask<E: Event>(&self, event: &E) -> Option<Vec<bool>> {
661 if !self.cross_rule_ac_enabled || self.cross_rule_ac_index.is_empty() {
662 return None;
663 }
664 let mut hits = vec![false; self.rules.len()];
665 self.cross_rule_ac_index.mark_hits(event, &mut hits);
666 // Compose: keep = !ac_prunable OR ac_hit. The prunable vector and
667 // the rule slice are kept aligned by `rebuild_index`.
668 for (idx, slot) in hits.iter_mut().enumerate() {
669 if !self
670 .cross_rule_ac_prunable
671 .get(idx)
672 .copied()
673 .unwrap_or(false)
674 {
675 *slot = true;
676 }
677 }
678 Some(hits)
679 }
680
681 #[cfg(not(feature = "daachorse-index"))]
682 #[inline(always)]
683 fn cross_rule_ac_keep_mask<E: Event>(&self, _event: &E) -> Option<Vec<bool>> {
684 None
685 }
686
687 /// Pick the candidate rule set for `event`. When a logsource extractor
688 /// produced an event logsource, the product-partitioned index drops
689 /// always-evaluated rules of a conflicting product; otherwise the full
690 /// candidate set is returned (zero behaviour change when pruning is off).
691 fn logsource_candidates<E: Event>(
692 &self,
693 event: &E,
694 event_logsource: Option<&LogSource>,
695 ) -> Vec<usize> {
696 match event_logsource {
697 Some(ls) => {
698 // Observability: count the fail-open case (no logsource at all)
699 // and the always-evaluated rules pruned by product conflict.
700 if ls.product.is_none() && ls.service.is_none() && ls.category.is_none() {
701 self.logsource_absent.fetch_add(1, Ordering::Relaxed);
702 }
703 let pruned = self
704 .rule_index
705 .conflicting_unindexable_count(ls.product.as_deref());
706 if pruned > 0 {
707 self.logsource_pruned
708 .fetch_add(pruned as u64, Ordering::Relaxed);
709 }
710 self.rule_index
711 .candidates_with_logsource(event, ls.product.as_deref())
712 }
713 None => self.rule_index.candidates(event),
714 }
715 }
716
717 fn evaluate_no_bloom_path<E: Event>(
718 &self,
719 event: &E,
720 event_logsource: Option<&LogSource>,
721 ) -> Vec<EvaluationResult> {
722 // Pass the zero-sized `NoBloom` lookup so this monomorphizes to the
723 // same straight-line code as the pre-bloom engine while still
724 // threading the configured match-detail level.
725 let keep = self.cross_rule_ac_keep_mask(event);
726 // `event_logsource` is `None` (the default) unless pruning is enabled,
727 // leaving the loop's behaviour unchanged.
728 let candidates = self.logsource_candidates(event, event_logsource);
729 let mut results = Vec::new();
730 for idx in candidates {
731 if let Some(ref mask) = keep
732 && !mask[idx]
733 {
734 continue;
735 }
736 let rule = &self.rules[idx];
737 if let Some(event_ls) = event_logsource
738 && !logsource_compatible(&rule.logsource, event_ls)
739 {
740 continue;
741 }
742 if let Some(mut m) =
743 evaluate_rule_with_bloom(rule, event, &bloom_index::NoBloom, self.match_detail)
744 {
745 if self.include_event
746 && let Some(d) = m.as_detection_mut()
747 && d.event.is_none()
748 {
749 d.event = Some(event.to_json());
750 }
751 results.push(m);
752 }
753 }
754 results
755 }
756
757 fn evaluate_with_bloom_path<E: Event>(
758 &self,
759 event: &E,
760 event_logsource: Option<&LogSource>,
761 ) -> Vec<EvaluationResult> {
762 let bloom = BloomCache::new(&self.bloom_index, event);
763 let keep = self.cross_rule_ac_keep_mask(event);
764 // `event_logsource` is `None` (the default) unless pruning is enabled,
765 // leaving the loop's behaviour unchanged.
766 let candidates = self.logsource_candidates(event, event_logsource);
767 let mut results = Vec::new();
768 for idx in candidates {
769 if let Some(ref mask) = keep
770 && !mask[idx]
771 {
772 continue;
773 }
774 let rule = &self.rules[idx];
775 if let Some(event_ls) = event_logsource
776 && !logsource_compatible(&rule.logsource, event_ls)
777 {
778 continue;
779 }
780 if let Some(mut m) = evaluate_rule_with_bloom(rule, event, &bloom, self.match_detail) {
781 if self.include_event
782 && let Some(d) = m.as_detection_mut()
783 && d.event.is_none()
784 {
785 d.event = Some(event.to_json());
786 }
787 results.push(m);
788 }
789 }
790 results
791 }
792
793 /// Evaluate an event against candidate rules matching the given logsource.
794 ///
795 /// Uses the inverted index for candidate pre-filtering, then applies the
796 /// logsource constraint. Only rules whose logsource is compatible with
797 /// `event_logsource` are evaluated.
798 pub fn evaluate_with_logsource<E: Event>(
799 &self,
800 event: &E,
801 event_logsource: &LogSource,
802 ) -> Vec<EvaluationResult> {
803 if self.bloom_prefilter {
804 self.evaluate_with_logsource_with_bloom(event, event_logsource)
805 } else {
806 self.evaluate_with_logsource_no_bloom(event, event_logsource)
807 }
808 }
809
810 fn evaluate_with_logsource_no_bloom<E: Event>(
811 &self,
812 event: &E,
813 event_logsource: &LogSource,
814 ) -> Vec<EvaluationResult> {
815 let keep = self.cross_rule_ac_keep_mask(event);
816 let mut results = Vec::new();
817 for idx in self.rule_index.candidates(event) {
818 if let Some(ref mask) = keep
819 && !mask[idx]
820 {
821 continue;
822 }
823 let rule = &self.rules[idx];
824 if logsource_matches(&rule.logsource, event_logsource)
825 && let Some(mut m) =
826 evaluate_rule_with_bloom(rule, event, &bloom_index::NoBloom, self.match_detail)
827 {
828 if self.include_event
829 && let Some(d) = m.as_detection_mut()
830 && d.event.is_none()
831 {
832 d.event = Some(event.to_json());
833 }
834 results.push(m);
835 }
836 }
837 results
838 }
839
840 fn evaluate_with_logsource_with_bloom<E: Event>(
841 &self,
842 event: &E,
843 event_logsource: &LogSource,
844 ) -> Vec<EvaluationResult> {
845 let bloom = BloomCache::new(&self.bloom_index, event);
846 let keep = self.cross_rule_ac_keep_mask(event);
847 let mut results = Vec::new();
848 for idx in self.rule_index.candidates(event) {
849 if let Some(ref mask) = keep
850 && !mask[idx]
851 {
852 continue;
853 }
854 let rule = &self.rules[idx];
855 if logsource_matches(&rule.logsource, event_logsource)
856 && let Some(mut m) =
857 evaluate_rule_with_bloom(rule, event, &bloom, self.match_detail)
858 {
859 if self.include_event
860 && let Some(d) = m.as_detection_mut()
861 && d.event.is_none()
862 {
863 d.event = Some(event.to_json());
864 }
865 results.push(m);
866 }
867 }
868 results
869 }
870
871 /// Evaluate a batch of events, returning per-event match results.
872 ///
873 /// When the `parallel` feature is enabled, events are evaluated concurrently
874 /// using rayon's work-stealing thread pool. Otherwise, falls back to
875 /// sequential evaluation.
876 pub fn evaluate_batch<E: Event + Sync>(&self, events: &[&E]) -> Vec<Vec<EvaluationResult>> {
877 #[cfg(feature = "parallel")]
878 {
879 use rayon::prelude::*;
880 events.par_iter().map(|e| self.evaluate(e)).collect()
881 }
882 #[cfg(not(feature = "parallel"))]
883 {
884 events.iter().map(|e| self.evaluate(e)).collect()
885 }
886 }
887
888 /// Number of rules loaded in the engine.
889 pub fn rule_count(&self) -> usize {
890 self.rules.len()
891 }
892
893 /// Access the compiled rules.
894 pub fn rules(&self) -> &[CompiledRule] {
895 &self.rules
896 }
897}
898
899impl Default for Engine {
900 fn default() -> Self {
901 Self::new()
902 }
903}