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