[FR Build a limited compatible rule ndjson for older stacks (#2885)

detection_rules/main.py

@@ -11,21 +11,22 @@ import os
 import re
 import time
 from datetime import datetime
+from marshmallow_dataclass import class_schema
 from pathlib import Path
-from typing import Dict, Optional
+from semver import Version
+from typing import Dict, List, Optional
 from uuid import uuid4
 
 import click
 
-
 from .cli_utils import rule_prompt, multi_collection
 from .mappings import build_coverage_map, get_triggered_rules, print_converage_summary
 from .misc import add_client, client_error, nested_set, parse_config, load_current_package_version
-from .rule import TOMLRule, TOMLRuleContents
+from .rule import TOMLRule, TOMLRuleContents, QueryRuleData
 from .rule_formatter import toml_write
 from .rule_loader import RuleCollection
-from .schemas import all_versions, definitions
-from .utils import get_path, get_etc_path, clear_caches, load_dump, load_rule_contents
+from .schemas import all_versions, definitions, get_incompatible_fields
+from .utils import Ndjson, get_path, get_etc_path, clear_caches, load_dump, load_rule_contents
 
 RULES_DIR = get_path('rules')
 
@@ -113,6 +114,57 @@ def import_rules(input_file, directory):
         rule_prompt(rule_path, required_only=True, save=True, verbose=True, additional_required=['index'], **contents)
 
 
+@root.command('build-limited-rules')
+@click.option('--stack-version', type=click.Choice(all_versions()), required=True,
+              help='Version to downgrade to be compatible with the older instance of Kibana')
+@click.option('--output-file', '-o', type=click.Path(dir_okay=False, exists=False), required=True)
+def build_limited_rules(stack_version: str, output_file: str):
+    """
+    Import rules from json, toml, or Kibana exported rule file(s),
+    filter out unsupported ones, and write to output NDJSON file.
+    """
+
+    # Schema generation and incompatible fields detection
+    query_rule_data = class_schema(QueryRuleData)()
+    fields = getattr(query_rule_data, 'fields', {})
+    incompatible_fields = get_incompatible_fields(list(fields.values()),
+                                                  Version.parse(stack_version, optional_minor_and_patch=True))
+
+    # Load all rules
+    rules = RuleCollection.default()
+
+    # Define output path
+    output_path = Path(output_file)
+
+    # Define ndjson instance for output
+    ndjson_output = Ndjson()
+
+    # Function to process each rule
+    def process_rule(rule, incompatible_fields: List[str]):
+        if rule.contents.type in definitions.UNSUPPORTED_RULE_TYPES:
+            click.secho(f'{rule.contents.name} - Skipping supported rule type: {rule.contents.get("type")}',
+                        fg='yellow')
+            return None
+
+        # Remove unsupported fields from rule
+        rule_contents = rule.contents.to_api_format()
+        for field in incompatible_fields:
+            rule_contents.pop(field, None)
+
+        return rule_contents
+
+    # Process each rule and add to ndjson_output
+    for rule in rules.rules:
+        processed_rule = process_rule(rule, incompatible_fields)
+        if processed_rule is not None:
+            ndjson_output.append(processed_rule)
+
+    # Write ndjson_output to file
+    ndjson_output.dump(output_path)
+
+    click.echo(f'Success: Rules written to {output_file}')
+
+
 @root.command('toml-lint')
 @click.option('--rule-file', '-f', multiple=True, type=click.Path(exists=True),
               help='Specify one or more rule files.')

detection_rules/schemas/__init__.py

@@ -15,11 +15,13 @@ from ..misc import load_current_package_version
 from ..utils import cached, get_etc_path, load_etc_dump
 from . import definitions
 from .rta_schema import validate_rta_mapping
+from .stack_compat import get_incompatible_fields
 
 __all__ = (
     "SCHEMA_DIR",
     "definitions",
     "downgrade",
+    "get_incompatible_fields",
     "get_min_supported_stack_version",
     "get_stack_schemas",
     "get_stack_versions",

detection_rules/schemas/definitions.py

@@ -36,6 +36,7 @@ MACHINE_LEARNING = 'machine_learning'
 SAVED_QUERY = 'saved_query'
 QUERY = 'query'
 QUERY_FIELD_OP_EXCEPTIONS = ["powershell.file.script_block_text"]
+UNSUPPORTED_RULE_TYPES = {'new_terms', 'threat_match', 'threshold', 'machine_learning'}
 
 # we had a bad rule ID make it in before tightening up the pattern, and so we have to let it bypass
 KNOWN_BAD_RULE_IDS = Literal['119c8877-8613-416d-a98a-96b6664ee73a5']