[FR] [DAC] Update default KQL parsing behavior to normalize keywords for custom rule directories.

detection_rules/config.py

@@ -189,6 +189,7 @@ class RulesConfig:
     bbr_rules_dirs: Optional[List[Path]] = field(default_factory=list)
     bypass_version_lock: bool = False
     exception_dir: Optional[Path] = None
+    normalize_kql_keywords: bool = True
 
     def __post_init__(self):
         """Perform post validation on packages.yaml file."""
@@ -271,6 +272,9 @@ def parse_rules_config(path: Optional[Path] = None) -> RulesConfig:
     if loaded.get('bbr_rules_dirs'):
         contents['bbr_rules_dirs'] = [base_dir.joinpath(d).resolve() for d in loaded.get('bbr_rules_dirs', [])]
 
+    # kql keyword normalization
+    contents['normalize_kql_keywords'] = loaded.get('normalize_kql_keywords', True)
+
     try:
         rules_config = RulesConfig(test_config=test_config, **contents)
     except (ValueError, TypeError) as e:

detection_rules/devtools.py

@@ -33,6 +33,7 @@
 from .beats import (download_beats_schema, download_latest_beats_schema,
                     refresh_main_schema)
 from .cli_utils import single_collection
+from .config import parse_rules_config
 from .docs import IntegrationSecurityDocs, IntegrationSecurityDocsMDX
 from .ecs import download_endpoint_schemas, download_schemas
 from .endgame import EndgameSchemaManager
@@ -50,7 +51,7 @@
                         Package)
 from .rule import (AnyRuleData, BaseRuleData, DeprecatedRule, QueryRuleData,
                    RuleTransform, ThreatMapping, TOMLRule, TOMLRuleContents)
-from .rule_loader import RULES_CONFIG, RuleCollection, production_filter
+from .rule_loader import RuleCollection, production_filter
 from .schemas import definitions, get_stack_versions
 from .utils import dict_hash, get_etc_path, get_path, load_dump
 from .version_lock import VersionLockFile, loaded_version_lock
@@ -61,6 +62,7 @@
 NAVIGATOR_BADGE = (
     f'[![ATT&CK navigator coverage](https://img.shields.io/badge/ATT&CK-Navigator-red.svg)]({NAVIGATOR_URL})'
 )
+RULES_CONFIG = parse_rules_config()
 
 
 def get_github_token() -> Optional[str]:

detection_rules/eswrap.py

@@ -17,6 +17,7 @@
 from elasticsearch.client import AsyncSearchClient
 
 import kql
+from .config import parse_rules_config
 from .main import root
 from .misc import add_params, client_error, elasticsearch_options, get_elasticsearch_client, nested_get
 from .rule import TOMLRule
@@ -26,6 +27,7 @@
 
 COLLECTION_DIR = get_path('collections')
 MATCH_ALL = {'bool': {'filter': [{'match_all': {}}]}}
+RULES_CONFIG = parse_rules_config()
 
 
 def add_range_to_dsl(dsl_filter, start_time, end_time='now'):
@@ -92,7 +94,7 @@ def evaluate_against_rule_and_update_mapping(self, rule_id, rta_name, verbose=Tr
         rule = RuleCollection.default().id_map.get(rule_id)
         assert rule is not None, f"Unable to find rule with ID {rule_id}"
         merged_events = combine_sources(*self.events.values())
-        filtered = evaluate(rule, merged_events)
+        filtered = evaluate(rule, merged_events, normalize_kql_keywords=RULES_CONFIG.normalize_kql_keywords)
 
         if filtered:
             sources = [e['agent']['type'] for e in filtered]

detection_rules/etc/_config.yaml

@@ -8,7 +8,7 @@ files:
   packages: packages.yaml
   stack_schema_map: stack-schema-map.yaml
   version_lock: version.lock.json
-
+normalize_kql_keywords: False
 # Set the versioning strategy.
 # 1. Set to False to use version.lock.json file
 # 2. Set to True to either:

detection_rules/rule.py

@@ -1526,8 +1526,8 @@ def get_unique_query_fields(rule: TOMLRule) -> List[str]:
 
         cfg = set_eql_config(rule.contents.metadata.get('min_stack_version'))
         with eql.parser.elasticsearch_syntax, eql.parser.ignore_missing_functions, eql.parser.skip_optimizations, cfg:
-            parsed = kql.parse(query) if language == 'kuery' else eql.parse_query(query)
-
+            parsed = (kql.parse(query, normalize_kql_keywords=RULES_CONFIG.normalize_kql_keywords)
+                      if language == 'kuery' else eql.parse_query(query))
         return sorted(set(str(f) for f in parsed if isinstance(f, (eql.ast.Field, kql.ast.Field))))
 
 

detection_rules/rule_validators.py

@@ -18,7 +18,7 @@
 import kql
 
 from . import ecs, endgame
-from .config import CUSTOM_RULES_DIR, load_current_package_version
+from .config import CUSTOM_RULES_DIR, load_current_package_version, parse_rules_config
 from .integrations import (get_integration_schema_data,
                            load_integrations_manifests)
 from .rule import (EQLRuleData, QueryRuleData, QueryValidator, RuleMeta,
@@ -33,6 +33,7 @@
                         eql.EqlSyntaxError,
                         eql.EqlTypeMismatchError]
 KQL_ERROR_TYPES = Union[kql.KqlCompileError, kql.KqlParseError]
+RULES_CONFIG = parse_rules_config()
 
 
 class ExtendedTypeHint(Enum):
@@ -107,7 +108,7 @@ class KQLValidator(QueryValidator):
 
     @cached_property
     def ast(self) -> kql.ast.Expression:
-        return kql.parse(self.query)
+        return kql.parse(self.query, normalize_kql_keywords=RULES_CONFIG.normalize_kql_keywords)
 
     @cached_property
     def unique_fields(self) -> List[str]:
@@ -151,7 +152,7 @@ def validate_stack_combos(self, data: QueryRuleData, meta: RuleMeta) -> Union[KQ
                                                                     beats_version, ecs_version)
 
             try:
-                kql.parse(self.query, schema=schema)
+                kql.parse(self.query, schema=schema, normalize_kql_keywords=RULES_CONFIG.normalize_kql_keywords)
             except kql.KqlParseError as exc:
                 message = exc.error_msg
                 trailer = err_trailer
@@ -212,7 +213,9 @@ def validate_integration(
 
             # Validate the query against the schema
             try:
-                kql.parse(self.query, schema=integration_schema)
+                kql.parse(self.query,
+                          schema=integration_schema,
+                          normalize_kql_keywords=RULES_CONFIG.normalize_kql_keywords)
             except kql.KqlParseError as exc:
                 if exc.error_msg == "Unknown field":
                     field = extract_error_field(self.query, exc)

detection_rules/utils.py

@@ -30,6 +30,7 @@
 
 import kql
 
+
 CURR_DIR = Path(__file__).resolve().parent
 ROOT_DIR = CURR_DIR.parent
 ETC_DIR = ROOT_DIR / "detection_rules" / "etc"
@@ -240,9 +241,9 @@ def convert_time_span(span: str) -> int:
     return eql.ast.TimeRange(amount, unit).as_milliseconds()
 
 
-def evaluate(rule, events):
+def evaluate(rule, events, normalize_kql_keywords: bool = False):
     """Evaluate a query against events."""
-    evaluator = kql.get_evaluator(kql.parse(rule.query))
+    evaluator = kql.get_evaluator(kql.parse(rule.query), normalize_kql_keywords=normalize_kql_keywords)
     filtered = list(filter(evaluator, events))
     return filtered
 

docs/custom-rules.md

@@ -113,6 +113,7 @@ class RulesConfig:
     bbr_rules_dirs: Optional[List[Path]] = field(default_factory=list)
     bypass_version_lock: bool = False
     exception_dir: Optional[Path] = None
+    normalize_kql_keywords: bool = True
 
 # using the stack_schema_map
 RULES_CONFIG.stack_schema_map

tests/test_all_rules.py

@@ -67,7 +67,7 @@ def test_all_rule_queries_optimized(self):
                 )
             ):
                 source = rule.contents.data.query
-                tree = kql.parse(source, optimize=False)
+                tree = kql.parse(source, optimize=False, normalize_kql_keywords=RULES_CONFIG.normalize_kql_keywords)
                 optimized = tree.optimize(recursive=True)
                 err_message = f'\n{self.rule_str(rule)} Query not optimized for rule\n' \
                               f'Expected: {optimized}\nActual: {source}'

tests/test_mappings.py

@@ -9,19 +9,23 @@
 import warnings
 
 from . import get_data_files, get_fp_data_files
+from detection_rules.config import parse_rules_config
 from detection_rules.utils import combine_sources, evaluate, load_etc_dump
 from rta import get_available_tests
 from .base import BaseRuleTest
 
 
+RULES_CONFIG = parse_rules_config()
+
+
 class TestMappings(BaseRuleTest):
     """Test that all rules appropriately match against expected data sets."""
 
     FP_FILES = get_fp_data_files()
 
     def evaluate(self, documents, rule, expected, msg):
         """KQL engine to evaluate."""
-        filtered = evaluate(rule, documents)
+        filtered = evaluate(rule, documents, RULES_CONFIG.normalize_kql_keywords)
         self.assertEqual(expected, len(filtered), msg)
         return filtered