[FR] Support new_terms schema import/export w/custom format

detection_rules/main.py

@@ -111,7 +111,10 @@ def import_rules_into_repo(input_file, required_only, directory):
         base_path = contents.get('name') or contents.get('rule', {}).get('name')
         base_path = rulename_to_filename(base_path) if base_path else base_path
         rule_path = os.path.join(RULES_DIR, base_path) if base_path else None
-        additional = ['index'] if not contents.get('data_view_id') else ['data_view_id']
+
+        # handle both rule json formats loaded from kibana and toml
+        data_view_id = contents.get("data_view_id") or contents.get("rule", {}).get("data_view_id")
+        additional = ["index"] if not data_view_id else ["data_view_id"]
         rule_prompt(rule_path, required_only=required_only, save=True, verbose=True,
                     additional_required=additional, **contents)
 

detection_rules/rule.py

@@ -21,7 +21,7 @@
 from semver import Version
 from marko.block import Document as MarkoDocument
 from marko.ext.gfm import gfm
-from marshmallow import ValidationError, validates_schema
+from marshmallow import ValidationError, validates_schema, pre_load
 
 import kql
 
@@ -768,6 +768,27 @@ class HistoryWindowStart:
     type: Literal["new_terms"]
     new_terms: NewTermsMapping
 
+    @pre_load
+    def preload_data(self, data: dict, **kwargs) -> dict:
+        """Preloads and formats the data to match the required schema."""
+        if "new_terms_fields" in data and "history_window_start" in data:
+            new_terms_mapping = {
+                "field": "new_terms_fields",
+                "value": data["new_terms_fields"],
+                "history_window_start": [
+                    {
+                        "field": "history_window_start",
+                        "value": data["history_window_start"]
+                    }
+                ]
+            }
+            data["new_terms"] = new_terms_mapping
+
+            # cleanup original fields after building into our toml format
+            data.pop("new_terms_fields")
+            data.pop("history_window_start")
+        return data
+
     def transform(self, obj: dict) -> dict:
         """Transforms new terms data to API format for Kibana."""
         obj[obj["new_terms"].get("field")] = obj["new_terms"].get("value")

detection_rules/rule_formatter.py

@@ -216,6 +216,11 @@ def _do_write(_data, _contents):
                 preserved_fields = ["params.message"]
                 v = [preserve_formatting_for_fields(action, preserved_fields) for action in v]
 
+            if k == 'filters':
+                # explicitly preserve formatting for value field in filters
+                preserved_fields = ["meta.value"]
+                v = [preserve_formatting_for_fields(meta, preserved_fields) for meta in v]
+
             if k == 'note' and isinstance(v, str):
                 # Transform instances of \ to \\ as calling write will convert \\ to \.
                 # This will ensure that the output file has the correct number of backslashes.