Eric Forte
72ba0b16a9
* add todo
* Add a normalize_kql_keywords function to utils
* update rule loader to normalize and warn
* optimized loading
* fix linting
* Moved conversion to kql module.
* Updated unit test
* Refactor KQL parser to normalize keywords via flag
* Fix logic typo
* Update detection_rules/utils.py
Co-authored-by: Justin Ibarra <16747370+brokensound77@users.noreply.github.com>
* Update lib/kql/kql/__init__.py
Co-authored-by: Justin Ibarra <16747370+brokensound77@users.noreply.github.com>
* Updated to fix unit tests and remove warnings
* linting typo
* Added comments
* remove unused imports
* Update kql.parse default
---------
Co-authored-by: Justin Ibarra <16747370+brokensound77@users.noreply.github.com>
Co-authored-by: Mika Ayenson <Mikaayenson@users.noreply.github.com>
(cherry picked from commit 1566c29bae)
81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
|
# or more contributor license agreements. Licensed under the Elastic License
|
|
# 2.0; you may not use this file except in compliance with the Elastic License
|
|
# 2.0.
|
|
|
|
import eql
|
|
|
|
from . import ast
|
|
from .dsl import ToDsl
|
|
from .eql2kql import Eql2Kql
|
|
from .errors import KqlParseError, KqlCompileError
|
|
from .evaluator import FilterGenerator
|
|
from .kql2eql import KqlToEQL
|
|
from .parser import lark_parse, KqlParser
|
|
|
|
__version__ = '0.1.6'
|
|
__all__ = (
|
|
"ast",
|
|
"from_eql",
|
|
"get_evaluator",
|
|
"KqlParseError",
|
|
"KqlCompileError",
|
|
"lint",
|
|
"parse",
|
|
"to_dsl",
|
|
"to_eql",
|
|
)
|
|
|
|
|
|
def to_dsl(parsed, optimize=True, schema=None):
|
|
"""Convert KQL to Elasticsearch Query DSL."""
|
|
if not isinstance(parsed, ast.KqlNode):
|
|
parsed = parse(parsed, optimize, schema)
|
|
|
|
return ToDsl.convert(parsed)
|
|
|
|
|
|
def to_eql(text, optimize=True, schema=None):
|
|
if isinstance(text, bytes):
|
|
text = text.decode("utf-8")
|
|
|
|
lark_parsed = lark_parse(text)
|
|
|
|
converted = KqlToEQL(text, schema=schema).visit(lark_parsed)
|
|
return converted.optimize(recursive=True) if optimize else converted
|
|
|
|
|
|
def parse(text, optimize: bool = True, schema: dict = None, normalize_kql_keywords: bool = False):
|
|
if isinstance(text, bytes):
|
|
text = text.decode("utf-8")
|
|
|
|
lark_parsed = lark_parse(text)
|
|
converted = KqlParser(text, schema=schema, normalize_kql_keywords=normalize_kql_keywords).visit(lark_parsed)
|
|
|
|
return converted.optimize(recursive=True) if optimize else converted
|
|
|
|
|
|
def lint(text):
|
|
if isinstance(text, bytes):
|
|
text = text.decode("utf-8")
|
|
|
|
return parse(text, optimize=True).render()
|
|
|
|
|
|
def from_eql(tree, optimize=True):
|
|
if not isinstance(tree, eql.ast.EqlNode):
|
|
try:
|
|
tree = eql.parse_query(tree, implied_any=True)
|
|
except eql.EqlSemanticError:
|
|
tree = eql.parse_expression(tree)
|
|
|
|
converted = Eql2Kql().walk(tree)
|
|
return converted.optimize(recursive=True) if optimize else converted
|
|
|
|
|
|
def get_evaluator(tree, optimize=False):
|
|
if not isinstance(tree, ast.KqlNode):
|
|
tree = parse(tree, optimize=optimize)
|
|
|
|
return FilterGenerator().filter(tree)
|