Skip to main content

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}