Support Ordered Temporal Proximity correlation

CHANGELOG-Japanese.md

@@ -8,6 +8,7 @@
 - `expand`修飾子が入っているルールで使用されるプレースホルダー名を出力する`expand-list`コマンドを追加した。(#1513) (@fukuseket)
 - `expand`フィールド修飾子に対応した。 (#1434) (@fukusuket)
 - Temporal Proximity（`temporal`）の相関ルールに対応した。 (#1446) (@fukusuket)
+- Temporal Ordered Proximity (`temporal_ordered`) の相関ルールに対応した。 (#1447) (@fukusuket)
 
 **改善:**
 

CHANGELOG.md

@@ -8,6 +8,7 @@
 - New `expand-list` command to output placeholder names used for rules with the `expand` modifier. (#1513) (@fukuseket)
 - Support for `expand` field modifiers. (#1434) (@fukusuket)
 - Suppport for Temporal Proximity (`temporal`) correlation rules. (#1446) (@fukusuket)
+- Suppport for Temporal Ordered Proximity (`temporal_ordered`) correlation rules. (#1447) (@fukusuket)
 
 **Enhancements:**
 

src/detections/detection.rs

@@ -199,25 +199,43 @@ impl Detection {
 
     fn detect_within_timeframe(
         ids: &[String],
-        data: &HashMap<String, Vec<AggResult>>,
+        temporal_ref_all_results: &HashMap<String, Vec<AggResult>>,
         timeframe: Duration,
+        temporal_ordered: bool,
     ) -> Vec<AggResult> {
         let mut result = Vec::new();
-        let key = ids[0].clone();
-        for y in data.get(key.as_str()).unwrap() {
-            let mut found = true;
-            for id in ids.iter().skip(1) {
-                if !data.get(id.as_str()).unwrap().iter().any(|t| {
-                    (t.start_timedate >= y.start_timedate - timeframe)
-                        && (t.start_timedate <= y.start_timedate + timeframe)
-                }) {
-                    found = false;
-                    break;
+        let key = ids.first();
+        if let Some(key) = key {
+            if let Some(base_records) = temporal_ref_all_results.get(key.as_str()) {
+                for base in base_records {
+                    let mut found = false;
+                    let mut last_base = base;
+                    for id in ids.iter().skip(1) {
+                        found = false;
+                        if let Some(target_records) = temporal_ref_all_results.get(id.as_str()) {
+                            if temporal_ordered {
+                                found = target_records.iter().any(|t| {
+                                    (t.start_timedate >= last_base.start_timedate)
+                                        && (t.start_timedate
+                                            <= last_base.start_timedate + timeframe)
+                                });
+                            } else {
+                                found = target_records.iter().any(|t| {
+                                    (t.start_timedate >= base.start_timedate - timeframe)
+                                        && (t.start_timedate <= base.start_timedate + timeframe)
+                                });
+                            }
+                            if !found {
+                                break;
+                            }
+                            last_base = base;
+                        }
+                    }
+                    if found {
+                        result.push(base.clone());
+                    }
                 }
             }
-            if found {
-                result.push(y.clone());
-            }
         }
         result
     }
@@ -230,35 +248,46 @@ impl Detection {
                 continue;
             }
             for value in rule.judge_satisfy_aggcondition(stored_static) {
-                if let CorrelationType::TemporalRef(_, uuid) = &rule.correlation_type {
+                let mut output = true;
+                if let CorrelationType::TemporalRef(generate, uuid) = &rule.correlation_type {
                     detected_temporal_refs
                         .entry(uuid.clone())
                         .or_insert_with(Vec::new)
                         .push(value.clone());
-                } else {
+                    output = *generate;
+                }
+                if output {
                     ret.push(Detection::create_agg_log_record(rule, value, stored_static));
                 }
             }
         }
         // temporalルールは個々ルールの判定がすべて出揃ってから判定できるため、再度rulesをループしてtemporalルールの判定を行う
         for rule in self.rules.iter() {
-            if let CorrelationType::Temporal(ref_ids) = &rule.correlation_type {
-                if ref_ids
-                    .iter()
-                    .all(|x| detected_temporal_refs.contains_key(x))
-                {
-                    let mut data = HashMap::new();
-                    for id in ref_ids {
-                        let entry = detected_temporal_refs.get_key_value(id);
-                        data.insert(entry.unwrap().0.clone(), entry.unwrap().1.clone());
-                    }
-                    let timeframe = get_sec_timeframe(rule, stored_static);
-                    if let Some(timeframe) = timeframe {
-                        let duration = Duration::seconds(timeframe);
-                        let values = Detection::detect_within_timeframe(ref_ids, &data, duration);
-                        for v in values {
-                            ret.push(Detection::create_agg_log_record(rule, v, stored_static));
-                        }
+            let (ref_ids, temporal_ordered) = match &rule.correlation_type {
+                CorrelationType::Temporal(ref_ids) => (ref_ids, false),
+                CorrelationType::TemporalOrdered(ref_ids) => (ref_ids, true),
+                _ => continue,
+            };
+            if ref_ids
+                .iter()
+                .all(|x| detected_temporal_refs.contains_key(x))
+            {
+                let mut data = HashMap::new();
+                for id in ref_ids {
+                    let entry = detected_temporal_refs.get_key_value(id);
+                    data.insert(entry.unwrap().0.clone(), entry.unwrap().1.clone());
+                }
+                let timeframe = get_sec_timeframe(rule, stored_static);
+                if let Some(timeframe) = timeframe {
+                    let duration = Duration::seconds(timeframe);
+                    let results = Detection::detect_within_timeframe(
+                        ref_ids,
+                        &data,
+                        duration,
+                        temporal_ordered,
+                    );
+                    for res in results {
+                        ret.push(Detection::create_agg_log_record(rule, res, stored_static));
                     }
                 }
             }

src/detections/rule/correlation_parser.rs

@@ -251,8 +251,9 @@ fn merge_referenced_rule(
     if rule_type != Some("event_count")
         && rule_type != Some("value_count")
         && rule_type != Some("temporal")
+        && rule_type != Some("temporal_ordered")
     {
-        let m = "The type of correlation rule only supports event_count/value_count/temporal.";
+        let m = "The type of correlation rule only supports event_count/value_count/temporal/temporal_ordered.";
         error_log(&rule.rulepath, m, stored_static, parse_error_count);
         return rule;
     }
@@ -279,7 +280,7 @@ fn merge_referenced_rule(
         error_log(&rule.rulepath, m, stored_static, parse_error_count);
         return rule;
     }
-    if rule_type == Some("temporal") {
+    if rule_type == Some("temporal") || rule_type == Some("temporal_ordered") {
         return rule;
     }
     let (referenced_rules, name_to_selection) =
@@ -345,24 +346,26 @@ fn parse_temporal_rules(
         let mut temporal_ref_ids: Vec<Yaml> = Vec::new();
         if let Some(ref_ids) = temporal_yaml["correlation"]["rules"].as_vec() {
             for ref_id in ref_ids {
-                for other_rule in other_rules.iter() {
-                    if is_referenced_rule(other_rule, ref_id.as_str().unwrap_or_default()) {
-                        let new_id = Uuid::new_v4();
-                        temporal_ref_ids.push(Yaml::String(new_id.to_string()));
+                for other_rule in other_rules.iter_mut() {
+                    let ref_id = ref_id.as_str().unwrap_or_default();
+                    if is_referenced_rule(other_rule, ref_id) {
+                        let generate = temporal_yaml["correlation"]["generate"]
+                            .as_bool()
+                            .unwrap_or_default();
                         let mut new_yaml = other_rule.yaml.clone();
+                        if other_rule.correlation_type != CorrelationType::None {
+                            other_rule.correlation_type =
+                                CorrelationType::TemporalRef(generate, ref_id.to_string());
+                            temporal_ref_ids.push(Yaml::String(ref_id.to_string()));
+                            continue;
+                        }
+                        let new_id = Uuid::new_v4().to_string();
                         if let Some(hash) = new_yaml.as_mut_hash() {
                             hash.insert(
                                 Yaml::String("id".to_string()),
                                 Yaml::String(new_id.to_string()),
                             );
                         }
-                        let generate = temporal_yaml["correlation"]["generate"]
-                            .as_bool()
-                            .unwrap_or_default();
-                        if !generate {
-                            referenced_del_ids
-                                .insert(ref_id.as_str().unwrap_or_default().to_string());
-                        }
                         let mut node = RuleNode::new(other_rule.rulepath.clone(), new_yaml);
                         let _ = node.init(stored_static);
                         node.correlation_type =
@@ -383,6 +386,10 @@ fn parse_temporal_rules(
                         detection.aggregation_condition = Some(agg_info);
                         node.detection = detection;
                         temporal_ref_rules.push(node);
+                        temporal_ref_ids.push(Yaml::String(new_id.to_string()));
+                        if !generate {
+                            referenced_del_ids.insert(ref_id.to_string());
+                        }
                     }
                 }
             }
@@ -422,10 +429,12 @@ pub fn parse_correlation_rules(
     let (correlation_rules, mut not_correlation_rules): (Vec<RuleNode>, Vec<RuleNode>) = rule_nodes
         .into_iter()
         .partition(|rule_node| !rule_node.yaml["correlation"].is_badvalue());
-    let (temporal_rules, not_temporal_rules): (Vec<RuleNode>, Vec<RuleNode>) = correlation_rules
-        .into_iter()
-        .partition(|rule_node| rule_node.yaml["correlation"]["type"].as_str() == Some("temporal"));
-    let mut correlation_parsed_rules: Vec<RuleNode> = not_temporal_rules
+    let (temporal_rules, not_temporal_rules): (Vec<RuleNode>, Vec<RuleNode>) =
+        correlation_rules.into_iter().partition(|rule_node| {
+            rule_node.yaml["correlation"]["type"].as_str() == Some("temporal")
+                || rule_node.yaml["correlation"]["type"].as_str() == Some("temporal_ordered")
+        });
+    let mut parsed_rules: Vec<RuleNode> = not_temporal_rules
         .into_iter()
         .map(|correlation_rule_node| {
             merge_referenced_rule(
@@ -436,11 +445,11 @@ pub fn parse_correlation_rules(
             )
         })
         .collect();
+    parsed_rules.extend(not_correlation_rules);
     let parsed_temporal_rules =
-        parse_temporal_rules(temporal_rules, &mut not_correlation_rules, stored_static);
-    correlation_parsed_rules.extend(not_correlation_rules);
-    correlation_parsed_rules.extend(parsed_temporal_rules);
-    correlation_parsed_rules
+        parse_temporal_rules(temporal_rules, &mut parsed_rules, stored_static);
+    parsed_rules.extend(parsed_temporal_rules);
+    parsed_rules
 }
 
 #[cfg(test)]

src/detections/rule/mod.rs

@@ -32,6 +32,7 @@ pub enum CorrelationType {
     EventCount,
     ValueCount,
     Temporal(Vec<String>),
+    TemporalOrdered(Vec<String>),
     TemporalRef(bool, String),
 }
 
@@ -44,14 +45,18 @@ impl CorrelationType {
         match correlation_type {
             "event_count" => CorrelationType::EventCount,
             "value_count" => CorrelationType::ValueCount,
-            "temporal" => {
+            "temporal" | "temporal_ordered" => {
                 let rules: Vec<String> = yaml["correlation"]["rules"]
                     .as_vec()
                     .unwrap()
                     .iter()
                     .map(|rule| rule.as_str().unwrap().to_string())
                     .collect();
-                CorrelationType::Temporal(rules)
+                if correlation_type == "temporal" {
+                    CorrelationType::Temporal(rules)
+                } else {
+                    CorrelationType::TemporalOrdered(rules)
+                }
             }
             _ => CorrelationType::None,
         }