use std::collections::HashMap;
use rsigma_parser::{LogSource, SigmaCollection, SigmaRule};
use crate::correlation_engine::{
CorrelationConfig, CorrelationEngine, CorrelationSnapshot, CorrelationStateSnapshot,
ProcessResult,
};
use crate::engine::Engine;
use crate::error::Result;
use crate::event::{Event, MappedEvent};
use crate::logsource::LogSourceExtractor;
use crate::pipeline::Pipeline;
use crate::pipeline::transformations::Transformation;
use crate::result::EvaluationResult;
use crate::result::MatchDetailLevel;
use crate::schema::{OnUnknown, RouteDecision, RoutingPlan, SchemaClassifier};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SchemaPruning {
pub schema: String,
pub eligible: usize,
pub pruned: usize,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RouteOutcome {
Evaluated,
EvaluatedUnknown,
Dropped,
Errored,
}
pub struct RouteResult {
pub results: ProcessResult,
pub schema: Option<String>,
pub outcome: RouteOutcome,
}
fn collect_field_map(pipelines: &[Pipeline]) -> HashMap<String, Vec<String>> {
let mut map: HashMap<String, Vec<String>> = HashMap::new();
for pipeline in pipelines {
for item in &pipeline.transformations {
if let Transformation::FieldNameMapping { mapping } = &item.transformation {
for (from, to) in mapping {
map.entry(from.clone())
.or_default()
.extend(to.iter().cloned());
}
}
}
}
map
}
enum Routed1 {
Skip,
Eval {
set: usize,
detections: Vec<EvaluationResult>,
},
}
fn rule_product_kept(rule: &SigmaRule, products: &std::collections::HashSet<String>) -> bool {
match &rule.logsource.product {
None => true,
Some(p) => products.contains(&p.to_ascii_lowercase()),
}
}
fn pipeline_changes_product(pipeline: &Pipeline) -> bool {
pipeline.transformations.iter().any(|item| {
matches!(
&item.transformation,
Transformation::ChangeLogsource {
product: Some(_),
..
}
)
})
}
fn resolve_event_logsource<E: Event>(
extractor: &LogSourceExtractor,
implied: Option<&LogSource>,
event: &E,
) -> LogSource {
let mut ls = extractor.extract(event);
if let Some(implied) = implied {
if ls.product.is_none() {
ls.product = implied.product.clone();
}
if ls.service.is_none() {
ls.service = implied.service.clone();
}
if ls.category.is_none() {
ls.category = implied.category.clone();
}
for (key, value) in &implied.custom {
ls.custom
.entry(key.clone())
.or_insert_with(|| value.clone());
}
}
ls
}
fn detect_one<E: Event>(
classifier: &SchemaClassifier,
plan: &RoutingPlan,
engines: &[Engine],
extractor: Option<&LogSourceExtractor>,
event: &E,
) -> Routed1 {
let schema = classifier.classify(event).map(|m| m.name);
match plan.decide(schema.as_deref()) {
RouteDecision::Drop | RouteDecision::Error => Routed1::Skip,
RouteDecision::Evaluate { set, .. } => {
let detections = match extractor {
Some(ex) => {
let implied = schema.as_deref().and_then(|s| plan.schema_logsource(s));
let ls = resolve_event_logsource(ex, implied, event);
engines[set].evaluate_pruned(event, &ls)
}
None => engines[set].evaluate(event),
};
Routed1::Eval { set, detections }
}
}
}
pub struct SchemaRouter {
classifier: SchemaClassifier,
plan: RoutingPlan,
engines: Vec<Engine>,
field_maps: Vec<HashMap<String, Vec<String>>>,
correlation: Option<CorrelationEngine>,
logsource_extractor: Option<LogSourceExtractor>,
}
impl SchemaRouter {
#[allow(clippy::too_many_arguments)]
pub fn build(
collection: &SigmaCollection,
classifier: SchemaClassifier,
plan: RoutingPlan,
pipeline_sets: Vec<Vec<Pipeline>>,
corr_config: CorrelationConfig,
include_event: bool,
match_detail: MatchDetailLevel,
logsource_extractor: Option<LogSourceExtractor>,
partition_rules: bool,
) -> Result<Self> {
let partition = if partition_rules {
plan.set_product_partition()
} else {
vec![None; pipeline_sets.len()]
};
let mut engines = Vec::with_capacity(pipeline_sets.len());
let mut field_maps = Vec::with_capacity(pipeline_sets.len());
for (idx, set) in pipeline_sets.iter().enumerate() {
let mut engine = Engine::new();
engine.set_include_event(include_event);
engine.set_match_detail(match_detail);
for p in set {
engine.add_pipeline(p.clone());
}
let partitioned = partition
.get(idx)
.and_then(|o| o.as_ref())
.filter(|_| !set.iter().any(pipeline_changes_product));
match partitioned {
Some(products) => {
let mut filtered = collection.clone();
filtered.rules.retain(|r| rule_product_kept(r, products));
engine.add_collection(&filtered)?;
}
None => engine.add_collection(collection)?,
}
engines.push(engine);
field_maps.push(collect_field_map(set));
}
let correlation = if collection.correlations.is_empty() {
None
} else {
let mut ce = CorrelationEngine::new(corr_config);
ce.set_include_event(include_event);
ce.set_match_detail(match_detail);
ce.add_collection(collection)?;
Some(ce)
};
Ok(SchemaRouter {
classifier,
plan,
engines,
field_maps,
correlation,
logsource_extractor,
})
}
pub fn on_unknown(&self) -> OnUnknown {
self.plan.on_unknown()
}
pub fn has_correlations(&self) -> bool {
self.correlation.is_some()
}
pub fn detection_rule_count(&self) -> usize {
self.engines.first().map(|e| e.rule_count()).unwrap_or(0)
}
pub fn engine_rule_counts(&self) -> Vec<usize> {
self.engines.iter().map(Engine::rule_count).collect()
}
pub fn logsource_pruned_total(&self) -> u64 {
self.engines
.iter()
.map(Engine::logsource_pruned_total)
.sum()
}
pub fn logsource_absent_total(&self) -> u64 {
self.engines
.iter()
.map(Engine::logsource_absent_total)
.sum()
}
pub fn schema_pruning_summary(&self) -> Vec<SchemaPruning> {
if self.logsource_extractor.is_none() {
return Vec::new();
}
let mut out = Vec::new();
for schema in self.plan.schemas_with_logsource() {
let Some(implied) = self.plan.schema_logsource(&schema) else {
continue;
};
let set = match self.plan.decide(Some(&schema)) {
RouteDecision::Evaluate { set, .. } => set,
RouteDecision::Drop | RouteDecision::Error => 0,
};
let (eligible, pruned) = self.engines[set].logsource_eligibility(implied);
out.push(SchemaPruning {
schema,
eligible,
pruned,
});
}
out.sort_by(|a, b| {
b.pruned
.cmp(&a.pruned)
.then_with(|| a.schema.cmp(&b.schema))
});
out
}
pub fn correlation_rule_count(&self) -> usize {
self.correlation
.as_ref()
.map(|c| c.correlation_rule_count())
.unwrap_or(0)
}
pub fn state_count(&self) -> usize {
self.correlation
.as_ref()
.map(|c| c.state_count())
.unwrap_or(0)
}
pub fn correlation_introspect(
&self,
id: Option<&str>,
group: Option<&str>,
) -> Option<CorrelationStateSnapshot> {
self.correlation
.as_ref()
.map(|c| c.introspect_filtered(id, group))
}
pub fn export_state(&self) -> Option<CorrelationSnapshot> {
self.correlation.as_ref().map(|c| c.export_state())
}
pub fn import_state(&mut self, snapshot: CorrelationSnapshot) -> bool {
match &mut self.correlation {
Some(c) => c.import_state(snapshot),
None => true,
}
}
pub fn process_batch<E: Event + Sync>(&mut self, events: &[&E]) -> Vec<ProcessResult> {
let classifier = &self.classifier;
let plan = &self.plan;
let engines = &self.engines;
let extractor = self.logsource_extractor.as_ref();
let phase1: Vec<Routed1> = {
#[cfg(feature = "parallel")]
{
use rayon::prelude::*;
events
.par_iter()
.map(|e| detect_one(classifier, plan, engines, extractor, *e))
.collect()
}
#[cfg(not(feature = "parallel"))]
{
events
.iter()
.map(|e| detect_one(classifier, plan, engines, extractor, *e))
.collect()
}
};
let field_maps = &self.field_maps;
let correlation = &mut self.correlation;
phase1
.into_iter()
.zip(events)
.map(|(routed, event)| match routed {
Routed1::Skip => Vec::new(),
Routed1::Eval { set, detections } => match correlation {
Some(ce) => {
let mapped = MappedEvent::new(*event, &field_maps[set]);
ce.correlate_detections(&mapped, detections)
}
None => detections,
},
})
.collect()
}
pub fn route(&mut self, event: &impl Event) -> RouteResult {
let schema = self.classifier.classify(event).map(|m| m.name);
match self.plan.decide(schema.as_deref()) {
RouteDecision::Drop => RouteResult {
results: Vec::new(),
schema,
outcome: RouteOutcome::Dropped,
},
RouteDecision::Error => RouteResult {
results: Vec::new(),
schema,
outcome: RouteOutcome::Errored,
},
RouteDecision::Evaluate { set, unknown } => {
let detections = match self.logsource_extractor.as_ref() {
Some(ex) => {
let implied = schema
.as_deref()
.and_then(|s| self.plan.schema_logsource(s));
let ls = resolve_event_logsource(ex, implied, event);
self.engines[set].evaluate_pruned(event, &ls)
}
None => self.engines[set].evaluate(event),
};
let results = match &mut self.correlation {
Some(ce) => {
let mapped = MappedEvent::new(event, &self.field_maps[set]);
ce.correlate_detections(&mapped, detections)
}
None => detections,
};
RouteResult {
results,
schema,
outcome: if unknown {
RouteOutcome::EvaluatedUnknown
} else {
RouteOutcome::Evaluated
},
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::JsonEvent;
use crate::pipeline::parse_pipeline;
use crate::schema::RoutingConfig;
use rsigma_parser::parse_sigma_yaml;
use serde_json::json;
const RULES: &str = r#"
title: Whoami
id: rule-whoami
logsource:
category: process_creation
product: windows
detection:
selection:
CommandLine|contains: whoami
condition: selection
level: high
"#;
const ECS_PIPELINE: &str = r#"
name: ecs_test
priority: 20
transformations:
- id: map
type: field_name_mapping
mapping:
CommandLine: process.command_line
User: user.name
"#;
fn plan(bindings: &[(&str, &[&str])]) -> RoutingPlan {
let config = RoutingConfig {
on_unknown: OnUnknown::Warn,
default_pipelines: vec![],
aliases: std::collections::HashMap::new(),
bindings: bindings
.iter()
.map(|(s, ps)| crate::schema::SchemaBinding {
schema: (*s).to_string(),
pipelines: ps.iter().map(|p| (*p).to_string()).collect(),
logsource: None,
})
.collect(),
};
RoutingPlan::from_config(&config)
}
#[test]
fn routes_ecs_event_to_ecs_engine() {
let collection = parse_sigma_yaml(RULES).unwrap();
let ecs = parse_pipeline(ECS_PIPELINE).unwrap();
let plan = plan(&[("ecs", &["ecs_test"])]);
let mut router = SchemaRouter::build(
&collection,
SchemaClassifier::builtin(),
plan,
vec![vec![], vec![ecs]],
CorrelationConfig::default(),
false,
MatchDetailLevel::Off,
None,
false,
)
.unwrap();
let ecs_event = json!({"ecs.version": "8.0.0", "process.command_line": "cmd /c whoami"});
let r = router.route(&JsonEvent::borrow(&ecs_event));
assert_eq!(r.schema.as_deref(), Some("ecs"));
assert_eq!(r.outcome, RouteOutcome::Evaluated);
assert_eq!(r.results.len(), 1, "ECS event matches via the ECS engine");
let native = json!({"CommandLine": "cmd /c whoami"});
let r = router.route(&JsonEvent::borrow(&native));
assert_eq!(r.schema.as_deref(), Some("generic_json"));
assert_eq!(r.results.len(), 1);
}
#[test]
fn cross_schema_correlation_groups_the_same_entity() {
let rules = r#"
title: Whoami
id: rule-whoami
logsource:
category: process_creation
product: windows
detection:
selection:
CommandLine|contains: whoami
condition: selection
level: high
---
title: Repeated whoami by user
correlation:
type: event_count
rules:
- rule-whoami
group-by:
- User
timespan: 1h
condition:
gte: 2
level: high
"#;
let collection = parse_sigma_yaml(rules).unwrap();
let ecs = parse_pipeline(ECS_PIPELINE).unwrap();
let plan = plan(&[("ecs", &["ecs_test"])]);
let config = CorrelationConfig {
timestamp_fallback: crate::correlation_engine::TimestampFallback::WallClock,
..Default::default()
};
let mut router = SchemaRouter::build(
&collection,
SchemaClassifier::builtin(),
plan,
vec![vec![], vec![ecs]],
config,
false,
MatchDetailLevel::Off,
None,
false,
)
.unwrap();
let ecs_event = json!({
"ecs.version": "8.0.0",
"process.command_line": "cmd /c whoami",
"user.name": "alice"
});
let r1 = router.route(&JsonEvent::borrow(&ecs_event));
assert_eq!(r1.schema.as_deref(), Some("ecs"));
assert!(
!r1.results.iter().any(|r| r.is_correlation()),
"first event must not fire the count>=2 correlation yet"
);
let native_event = json!({"CommandLine": "cmd /c whoami", "User": "alice"});
let r2 = router.route(&JsonEvent::borrow(&native_event));
assert!(
r2.results.iter().any(|r| r.is_correlation()),
"the two events share group User=alice across schemas and must correlate"
);
}
#[test]
fn drop_policy_skips_unknown_events() {
let collection = parse_sigma_yaml(RULES).unwrap();
let config = RoutingConfig {
on_unknown: OnUnknown::Drop,
default_pipelines: vec![],
aliases: std::collections::HashMap::new(),
bindings: vec![],
};
let plan = RoutingPlan::from_config(&config);
let mut router = SchemaRouter::build(
&collection,
SchemaClassifier::new(vec![]),
plan,
vec![vec![]],
CorrelationConfig::default(),
false,
MatchDetailLevel::Off,
None,
false,
)
.unwrap();
let native = json!({"CommandLine": "cmd /c whoami"});
let r = router.route(&JsonEvent::borrow(&native));
assert_eq!(r.schema, None);
assert_eq!(r.outcome, RouteOutcome::Dropped);
assert!(r.results.is_empty());
}
#[test]
fn schema_derived_logsource_prunes_cross_product_rules() {
let rules = r#"
title: Win whoami
id: win-whoami
logsource:
category: process_creation
product: windows
detection:
selection:
CommandLine|contains: whoami
condition: selection
level: high
---
title: Linux whoami
id: linux-whoami
logsource:
category: process_creation
product: linux
detection:
selection:
CommandLine|contains: whoami
condition: selection
level: high
"#;
let collection = parse_sigma_yaml(rules).unwrap();
let event = json!({
"EventID": 1,
"ProcessGuid": "{abc}",
"Image": "C:/Windows/System32/cmd.exe",
"CommandLine": "cmd /c whoami"
});
let mut plain = SchemaRouter::build(
&collection,
SchemaClassifier::builtin(),
plan(&[]),
vec![vec![]],
CorrelationConfig::default(),
false,
MatchDetailLevel::Off,
None,
false,
)
.unwrap();
let r = plain.route(&JsonEvent::borrow(&event));
assert_eq!(r.schema.as_deref(), Some("sysmon"));
assert_eq!(r.results.len(), 2, "no pruning without an extractor");
let mut pruned = SchemaRouter::build(
&collection,
SchemaClassifier::builtin(),
plan(&[]),
vec![vec![]],
CorrelationConfig::default(),
false,
MatchDetailLevel::Off,
Some(LogSourceExtractor::new()),
false,
)
.unwrap();
let r = pruned.route(&JsonEvent::borrow(&event));
assert_eq!(r.schema.as_deref(), Some("sysmon"));
assert_eq!(
r.results.len(),
1,
"schema-derived product prunes the Linux rule"
);
assert_eq!(pruned.logsource_pruned_total(), 1);
let summary = pruned.schema_pruning_summary();
let sysmon = summary
.iter()
.find(|s| s.schema == "sysmon")
.expect("sysmon in summary");
assert_eq!(sysmon.eligible, 1);
assert_eq!(sysmon.pruned, 1);
assert!(!summary.iter().any(|s| s.schema == "ecs"));
assert!(plain.schema_pruning_summary().is_empty());
}
#[test]
fn partition_rules_compiles_only_applicable_rules_per_set() {
let rules = r#"
title: Win whoami
id: win-whoami
logsource:
category: process_creation
product: windows
detection:
selection:
CommandLine|contains: whoami
condition: selection
level: high
---
title: Linux whoami
id: linux-whoami
logsource:
category: process_creation
product: linux
detection:
selection:
CommandLine|contains: whoami
condition: selection
level: high
---
title: Any whoami
id: any-whoami
logsource:
category: process_creation
detection:
selection:
CommandLine|contains: whoami
condition: selection
level: high
"#;
let collection = parse_sigma_yaml(rules).unwrap();
let passthrough = parse_pipeline(
"name: passthrough\npriority: 10\ntransformations:\n - id: noop\n type: field_name_mapping\n mapping:\n __unused_a: __unused_b\n",
)
.unwrap();
let plan = plan(&[("sysmon", &["passthrough"])]);
let router = SchemaRouter::build(
&collection,
SchemaClassifier::builtin(),
plan,
vec![vec![], vec![passthrough]],
CorrelationConfig::default(),
false,
MatchDetailLevel::Off,
None,
true, )
.unwrap();
assert_eq!(router.engine_rule_counts(), vec![3, 2]);
}
#[test]
fn partition_rules_off_keeps_full_ruleset() {
let rules = r#"
title: Win whoami
id: win-whoami
logsource:
product: windows
detection:
selection:
CommandLine|contains: whoami
condition: selection
---
title: Linux whoami
id: linux-whoami
logsource:
product: linux
detection:
selection:
CommandLine|contains: whoami
condition: selection
"#;
let collection = parse_sigma_yaml(rules).unwrap();
let passthrough = parse_pipeline(
"name: passthrough\npriority: 10\ntransformations:\n - id: noop\n type: field_name_mapping\n mapping:\n __unused_a: __unused_b\n",
)
.unwrap();
let plan = plan(&[("sysmon", &["passthrough"])]);
let router = SchemaRouter::build(
&collection,
SchemaClassifier::builtin(),
plan,
vec![vec![], vec![passthrough]],
CorrelationConfig::default(),
false,
MatchDetailLevel::Off,
None,
false, )
.unwrap();
assert_eq!(router.engine_rule_counts(), vec![2, 2]);
}
}