//! Integration tests for the `engine explain` subcommand.
mod common;
use common::{rsigma, temp_file};
use insta::assert_snapshot;
use predicates::prelude::*;
const RULE: &str = r#"
title: Suspicious PowerShell
id: ps-1
logsource:
category: process_creation
detection:
selection:
Image|endswith: '\powershell.exe'
CommandLine|contains: '-enc'
filter:
User: SYSTEM
condition: selection and not filter
level: high
"#;
fn rule_file() -> tempfile::NamedTempFile {
temp_file(".yml", RULE)
}
#[test]
fn explains_a_match_as_human_tree() {
let f = rule_file();
rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"Image":"C:\\Windows\\powershell.exe","CommandLine":"powershell -enc AAAA","User":"alice"}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("Suspicious PowerShell (ps-1): MATCH"))
.stdout(predicate::str::contains("PASS selection"));
}
#[test]
fn near_miss_reports_value_mismatch_with_actual() {
let f = rule_file();
rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"Image":"C:\\Windows\\cmd.exe","CommandLine":"powershell -enc AAAA","User":"SYSTEM"}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("NO MATCH"))
.stdout(predicate::str::contains("value mismatch"))
.stdout(predicate::str::contains("actual=\"C:\\\\Windows\\\\cmd.exe\""));
}
#[test]
fn absent_field_reports_field_absent() {
let f = rule_file();
rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"Image":"C:\\Windows\\powershell.exe"}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("field absent"));
}
#[test]
fn case_only_difference_reports_case_mismatch() {
let rule = temp_file(
".yml",
r#"
title: Cased
id: cased-1
logsource:
category: process_creation
detection:
selection:
CommandLine|endswith|cased: '\powershell.exe'
condition: selection
"#,
);
rsigma()
.args([
"engine",
"explain",
"-r",
rule.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"CommandLine":"C:\\Windows\\POWERSHELL.EXE"}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("case mismatch"));
}
#[test]
fn negation_node_is_rendered() {
let f = rule_file();
rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"Image":"C:\\Windows\\powershell.exe","CommandLine":"-enc","User":"SYSTEM"}"#,
])
.assert()
.success()
// filter matches (User=SYSTEM) so `not filter` is FAIL, no overall match.
.stdout(predicate::str::contains("not:"))
.stdout(predicate::str::contains("NO MATCH"));
}
#[test]
fn quantified_selector_reports_counts_in_explain() {
// `1 of selection_*` is preserved as a native selector, so the trace is a
// quantified node with need/got counts. Snapshot the raw stdout so key
// order comes from the struct `Serialize` impls, not serde_json's
// `preserve_order` feature (which is not always unified into the build).
let rule = temp_file(
".yml",
r#"
title: One Of
id: oneof-1
logsource:
category: test
detection:
selection_a:
CommandLine|contains: powershell
selection_b:
CommandLine|contains: whoami
condition: 1 of selection_*
"#,
);
let output = rsigma()
.args([
"engine",
"explain",
"-r",
rule.path().to_str().unwrap(),
"--color",
"never",
"--output-format",
"json",
"-e",
r#"{"CommandLine":"run powershell"}"#,
])
.output()
.unwrap();
assert!(
output.status.success(),
"stderr={}",
String::from_utf8_lossy(&output.stderr)
);
assert_snapshot!(
String::from_utf8_lossy(&output.stdout).trim(),
@r#"[{"rule_title":"One Of","rule_id":"oneof-1","matched":true,"conditions":[{"type":"quantified","quantifier":"any","matched":true,"need":1,"got":1,"branches":[{"name":"selection_a","matched":true,"detection":{"type":"all_of","matched":true,"items":[{"field":"CommandLine","matcher":"contains","pattern":"powershell","actual":"run powershell","matched":true,"reason":"matched"}]}},{"name":"selection_b","matched":false,"detection":{"type":"all_of","matched":false,"items":[{"field":"CommandLine","matcher":"contains","pattern":"whoami","actual":"run powershell","matched":false,"reason":"value_mismatch"}]}}]}]}]"#
);
}
#[test]
fn json_output_serializes_the_trace() {
let f = rule_file();
rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--output-format",
"json",
"-e",
r#"{"Image":"C:\\Windows\\cmd.exe"}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("\"matched\":false"))
.stdout(predicate::str::contains("\"reason\":\"value_mismatch\""))
.stdout(predicate::str::contains("\"type\":\"selection\""));
}
#[test]
fn pipeline_rename_is_reflected_in_the_trace() {
let rule = temp_file(
".yml",
r#"
title: CmdLine
id: cl-1
logsource:
category: process_creation
detection:
selection:
CommandLine|contains: mimikatz
condition: selection
"#,
);
let pipeline = temp_file(
".yml",
r#"
name: ecs-ish
priority: 10
transformations:
- type: field_name_mapping
mapping:
CommandLine: process.command_line
"#,
);
rsigma()
.args([
"engine",
"explain",
"-r",
rule.path().to_str().unwrap(),
"-p",
pipeline.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"process":{"command_line":"mimikatz.exe"}}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("process.command_line"))
.stdout(predicate::str::contains("MATCH"));
}
#[test]
fn rule_id_filter_selects_one_rule() {
let rule = temp_file(
".yml",
r#"
title: First
id: first
logsource: {category: test}
detection:
selection:
A: 1
condition: selection
---
title: Second
id: second
logsource: {category: test}
detection:
selection:
B: 2
condition: selection
"#,
);
rsigma()
.args([
"engine",
"explain",
"-r",
rule.path().to_str().unwrap(),
"--rule-id",
"second",
"--color",
"never",
"-e",
r#"{"B":2}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("Second (second): MATCH"))
.stdout(predicate::str::contains("First").not());
}
#[test]
fn invalid_inline_json_exits_nonzero() {
let f = rule_file();
rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"-e",
"not json",
])
.assert()
.failure()
.stderr(predicate::str::contains("invalid JSON event"));
}
const ARRAY_RULE: &str = r#"
title: Suspicious connection
id: arr-1
sigma-version: 3
logsource:
category: network_connection
detection:
selection:
connections[any]:
protocol: 'TCP'
ip|cidr: '123.1.0.0/16'
condition: selection
"#;
const NESTED_ARRAY_RULE: &str = r#"
title: Nested rules
id: nest-1
sigma-version: 3
logsource:
category: test
detection:
selection:
rules[any]:
type: 'allow'
ip[all]|startswith: '123.1.1'
condition: selection
"#;
#[test]
fn array_match_human_tree() {
let f = temp_file(".yml", ARRAY_RULE);
let output = rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"connections":[{"protocol":"UDP","ip":"10.0.0.1"},{"protocol":"TCP","ip":"123.1.9.9"}]}"#,
])
.output()
.unwrap();
assert!(
output.status.success(),
"stderr={}",
String::from_utf8_lossy(&output.stderr)
);
assert_snapshot!(String::from_utf8_lossy(&output.stdout).trim());
}
#[test]
fn array_miss_human_tree() {
let f = temp_file(".yml", ARRAY_RULE);
let output = rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"connections":[{"protocol":"TCP","ip":"10.0.0.1"},{"protocol":"UDP","ip":"123.1.9.9"}]}"#,
])
.output()
.unwrap();
assert!(
output.status.success(),
"stderr={}",
String::from_utf8_lossy(&output.stderr)
);
assert_snapshot!(String::from_utf8_lossy(&output.stdout).trim());
}
#[test]
fn nested_array_match_human_tree() {
let f = temp_file(".yml", NESTED_ARRAY_RULE);
let output = rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--color",
"never",
"-e",
r#"{"rules":[{"type":"allow","ip":["123.1.1.1","123.1.1.2"]}]}"#,
])
.output()
.unwrap();
assert!(
output.status.success(),
"stderr={}",
String::from_utf8_lossy(&output.stderr)
);
assert_snapshot!(String::from_utf8_lossy(&output.stdout).trim());
}
#[test]
fn array_match_csv_uses_indexed_paths() {
let f = temp_file(".yml", ARRAY_RULE);
rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--output-format",
"csv",
"-e",
r#"{"connections":[{"protocol":"UDP","ip":"10.0.0.1"},{"protocol":"TCP","ip":"123.1.9.9"}]}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("connections[0].protocol"))
.stdout(predicate::str::contains("connections[1].protocol"))
.stdout(predicate::str::contains("connections[1].ip"));
}
#[test]
fn nested_array_csv_uses_nested_indexed_paths() {
let f = temp_file(".yml", NESTED_ARRAY_RULE);
rsigma()
.args([
"engine",
"explain",
"-r",
f.path().to_str().unwrap(),
"--output-format",
"csv",
"-e",
r#"{"rules":[{"type":"allow","ip":["123.1.1.1","123.1.1.2"]}]}"#,
])
.assert()
.success()
.stdout(predicate::str::contains("rules[0].type"))
.stdout(predicate::str::contains("rules[0].ip[0]"))
.stdout(predicate::str::contains("rules[0].ip[1]"));
}