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}