[FR] Add support for dataviews in the rule schema

detection_rules/main.py

@@ -114,7 +114,8 @@ def name_to_filename(name):
         base_path = contents.get('name') or contents.get('rule', {}).get('name')
         base_path = name_to_filename(base_path) if base_path else base_path
         rule_path = os.path.join(RULES_DIR, base_path) if base_path else None
-        rule_prompt(rule_path, required_only=True, save=True, verbose=True, additional_required=['index'], **contents)
+        additional = ['index'] if not contents.get('data_view_id') else ['data_view_id']
+        rule_prompt(rule_path, required_only=True, save=True, verbose=True, additional_required=additional, **contents)
 
 
 @root.command('build-limited-rules')

detection_rules/rule.py

@@ -568,7 +568,7 @@ def get_beats_schema(self, index: list, beats_version: str, ecs_version: str) ->
     def get_endgame_schema(self, index: list, endgame_version: str) -> Optional[endgame.EndgameSchema]:
         """Get an assembled flat endgame schema."""
 
-        if "endgame-*" not in index:
+        if index and "endgame-*" not in index:
             return None
 
         endgame_schema = endgame.read_endgame_schema(endgame_version=endgame_version)
@@ -581,6 +581,7 @@ class QueryRuleData(BaseRuleData):
     type: Literal["query"]
 
     index: Optional[List[str]]
+    data_view_id: Optional[str]
     query: str
     language: definitions.FilterLanguages
     alert_suppression: Optional[AlertSuppressionMapping] = field(metadata=dict(metadata=dict(min_compat="8.8")))

detection_rules/rule_validators.py

@@ -263,7 +263,7 @@ def validate_stack_combos(self, data: QueryRuleData, meta: RuleMeta) -> Union[EQ
 
             beat_types, beat_schema, schema = self.get_beats_schema(data.index or [],
                                                                     beats_version, ecs_version)
-            endgame_schema = self.get_endgame_schema(data.index, endgame_version)
+            endgame_schema = self.get_endgame_schema(data.index or [], endgame_version)
             eql_schema = ecs.KqlSchema2Eql(schema)
 
             # validate query against the beats and eql schema
@@ -312,8 +312,9 @@ def validate_integration(self, data: QueryRuleData, meta: RuleMeta,
             stack_version = integration_schema_data["stack_version"]
 
             # add non-ecs-schema fields for edge cases not added to the integration
-            for index_name in data.index:
-                integration_schema.update(**ecs.flatten(ecs.get_index_schema(index_name)))
+            if data.index:
+                for index_name in data.index:
+                    integration_schema.update(**ecs.flatten(ecs.get_index_schema(index_name)))
 
             # add endpoint schema fields for multi-line fields
             integration_schema.update(**ecs.flatten(ecs.get_endpoint_schemas()))