Merge pull request #114 from UncoderIO/gis-7328

Created base platform: aql. And fixes for qradar

Author: nazargesyk

uncoder-core/app/translator/core/exceptions/core.py

@@ -77,3 +77,7 @@ class InvalidYamlStructure(InvalidRuleStructure):
 
 class InvalidJSONStructure(InvalidRuleStructure):
     rule_type: str = "JSON"
+
+
+class InvalidXMLStructure(InvalidRuleStructure):
+    rule_type: str = "XML"

uncoder-core/app/translator/core/mixins/rule.py

@@ -1,8 +1,10 @@
 import json
+from typing import Union
 
+import xmltodict
 import yaml
 
-from app.translator.core.exceptions.core import InvalidJSONStructure, InvalidYamlStructure
+from app.translator.core.exceptions.core import InvalidJSONStructure, InvalidXMLStructure, InvalidYamlStructure
 from app.translator.core.mitre import MitreConfig
 
 
@@ -36,5 +38,13 @@ def parse_mitre_attack(self, tags: list[str]) -> dict[str, list]:
                     result["techniques"].append(technique)
             elif tactic := self.mitre_config.get_tactic(tag):
                 result["tactics"].append(tactic)
-
         return result
+
+
+class XMLRuleMixin:
+    @staticmethod
+    def load_rule(text: Union[str, bytes]) -> dict:
+        try:
+            return xmltodict.parse(text)
+        except Exception as err:
+            raise InvalidXMLStructure(error=str(err)) from err

uncoder-core/app/translator/core/models/query_container.py

@@ -56,6 +56,13 @@ class RawQueryContainer:
     meta_info: MetaInfoContainer = field(default_factory=MetaInfoContainer)
 
 
+@dataclass
+class RawQueryDictContainer:
+    query: dict
+    language: str
+    meta_info: dict
+
+
 @dataclass
 class TokenizedQueryContainer:
     tokens: list[TOKEN_TYPE]

uncoder-core/app/translator/core/parser.py

@@ -32,9 +32,13 @@
 
 class QueryParser(ABC):
     wrapped_with_comment_pattern: str = None
+    details: PlatformDetails = None
 
     def remove_comments(self, text: str) -> str:
-        return re.sub(self.wrapped_with_comment_pattern, "\n", text, flags=re.MULTILINE).strip()
+        if self.wrapped_with_comment_pattern:
+            return re.sub(self.wrapped_with_comment_pattern, "\n", text, flags=re.MULTILINE).strip()
+
+        return text
 
     def parse_raw_query(self, text: str, language: str) -> RawQueryContainer:
         return RawQueryContainer(query=text, language=language)
@@ -47,7 +51,6 @@ def parse(self, raw_query_container: RawQueryContainer) -> TokenizedQueryContain
 class PlatformQueryParser(QueryParser, ABC):
     mappings: BasePlatformMappings = None
     tokenizer: QueryTokenizer = None
-    details: PlatformDetails = None
     platform_functions: PlatformFunctions = None
 
     def get_fields_tokens(self, tokens: list[Union[FieldValue, Keyword, Identifier]]) -> list[Field]:

uncoder-core/app/translator/core/render_cti.py

@@ -19,6 +19,7 @@
 
 
 from app.translator.core.models.iocs import IocsChunkValue
+from app.translator.core.models.platform_details import PlatformDetails
 
 
 class RenderCTI:
@@ -31,6 +32,7 @@ class RenderCTI:
     final_result_for_many: str = "union * | where ({result})\n"
     final_result_for_one: str = "union * | where {result}\n"
     default_mapping = None
+    details: PlatformDetails = None
 
     def create_field_value(self, field: str, value: str, generic_field: str) -> str:  # noqa: ARG002
         return self.field_value_template.format(key=field, value=value)

uncoder-core/app/translator/core/tokenizer.py

@@ -52,6 +52,8 @@ class QueryTokenizer(BaseTokenizer):
     single_value_operators_map: ClassVar[dict[str, str]] = {}
     # used to generate re pattern. so the keys order is important
     multi_value_operators_map: ClassVar[dict[str, str]] = {}
+    # used to generate re pattern. so the keys order is important
+    fields_operator_map: ClassVar[dict[str, str]] = {}
     operators_map: ClassVar[dict[str, str]] = {}  # used to generate re pattern. so the keys order is important
 
     logical_operator_pattern = r"^(?P<logical_operator>and|or|not|AND|OR|NOT)\s+"
@@ -73,7 +75,11 @@ class QueryTokenizer(BaseTokenizer):
     def __init_subclass__(cls, **kwargs):
         cls._validate_re_patterns()
         cls.value_pattern = cls.base_value_pattern.replace("___value_pattern___", cls._value_pattern)
-        cls.operators_map = {**cls.single_value_operators_map, **cls.multi_value_operators_map}
+        cls.operators_map = {
+            **cls.single_value_operators_map,
+            **cls.multi_value_operators_map,
+            **cls.fields_operator_map,
+        }
         cls.operator_pattern = rf"""(?:___field___\s*(?P<operator>(?:{'|'.join(cls.operators_map)})))\s*"""
 
     @classmethod

uncoder-core/app/translator/managers.py

@@ -1,21 +1,16 @@
 from abc import ABC
 from functools import cached_property
+from typing import ClassVar, Union
 
 from app.models.translation import TranslatorPlatform
-from app.translator.core.exceptions.core import UnsupportedRootAParser
+from app.translator.core.exceptions.core import UnsupportedPlatform, UnsupportedRootAParser
+from app.translator.core.parser import QueryParser
+from app.translator.core.render import QueryRender
+from app.translator.core.render_cti import RenderCTI
 
 
-class Manager(ABC):
-    platforms = {}
-
-    def register(self, cls):
-        self.platforms[cls.details.platform_id] = cls()
-        return cls
-
-    def get(self, platform_id: str):  # noqa: ANN201
-        if platform := self.platforms.get(platform_id):
-            return platform
-        raise UnsupportedRootAParser(parser=platform_id)
+class PlatformManager(ABC):
+    platforms: ClassVar[dict[str, Union[QueryParser, QueryRender, RenderCTI]]] = {}
 
     def all_platforms(self) -> list:
         return list(self.platforms.keys())
@@ -40,54 +35,61 @@ def get_platforms_details(self) -> list[TranslatorPlatform]:
         return sorted(platforms, key=lambda platform: platform.group_name)
 
 
-class ParserManager(Manager):
-    platforms = {}
-    supported_by_roota_platforms = {}
-    main_platforms = {}
+class ParserManager(PlatformManager):
+    supported_by_roota_platforms: ClassVar[dict[str, QueryParser]] = {}
+    main_platforms: ClassVar[dict[str, QueryParser]] = {}
 
-    def get_supported_by_roota(self, platform_id: str):  # noqa: ANN201
+    def get(self, platform_id: str) -> QueryParser:
+        if platform := self.platforms.get(platform_id):
+            return platform
+        raise UnsupportedPlatform(platform=platform_id, is_parser=True)
+
+    def register(self, cls: type[QueryParser]) -> type[QueryParser]:
+        self.platforms[cls.details.platform_id] = cls()
+        return cls
+
+    def get_supported_by_roota(self, platform_id: str) -> QueryParser:
         if platform := self.supported_by_roota_platforms.get(platform_id):
             return platform
         raise UnsupportedRootAParser(parser=platform_id)
 
-    def register_supported_by_roota(self, cls):
+    def register_supported_by_roota(self, cls: type[QueryParser]) -> type[QueryParser]:
         parser = cls()
         self.supported_by_roota_platforms[cls.details.platform_id] = parser
         self.platforms[cls.details.platform_id] = parser
         return cls
 
-    def register_main(self, cls):
+    def register_main(self, cls: type[QueryParser]) -> type[QueryParser]:
         parser = cls()
         self.main_platforms[cls.details.platform_id] = parser
         self.platforms[cls.details.platform_id] = parser
         return cls
 
-    @cached_property
-    def get_platforms_details(self) -> list[TranslatorPlatform]:
-        platforms = [
-            TranslatorPlatform(
-                id=platform.details.platform_id,
-                name=platform.details.name,
-                code=platform.details.platform_id,
-                group_name=platform.details.group_name,
-                group_id=platform.details.group_id,
-                platform_name=platform.details.platform_name,
-                platform_id=platform.details.platform_id,
-                alt_platform_name=platform.details.alt_platform_name,
-                alt_platform=platform.details.alt_platform,
-                first_choice=platform.details.first_choice,
-            )
-            for platform in self.platforms.values()
-        ]
-        return sorted(platforms, key=lambda platform: platform.group_name)
 
+class RenderManager(PlatformManager):
+    platforms: ClassVar[dict[str, QueryRender]] = {}
+
+    def get(self, platform_id: str) -> QueryRender:
+        if platform := self.platforms.get(platform_id):
+            return platform
+        raise UnsupportedPlatform(platform=platform_id)
+
+    def register(self, cls: type[QueryRender]) -> type[QueryRender]:
+        self.platforms[cls.details.platform_id] = cls()
+        return cls
 
-class RenderManager(Manager):
-    platforms = {}
 
+class RenderCTIManager(PlatformManager):
+    platforms: ClassVar[dict[str, RenderCTI]] = {}
 
-class RenderCTIManager(Manager):
-    platforms = {}
+    def get(self, platform_id: str) -> RenderCTI:
+        if platform := self.platforms.get(platform_id):
+            return platform
+        raise UnsupportedPlatform(platform=platform_id)
+
+    def register(self, cls: type[RenderCTI]) -> type[RenderCTI]:
+        self.platforms[cls.details.platform_id] = cls()
+        return cls
 
 
 parser_manager = ParserManager()

uncoder-core/app/translator/platforms/arcsight/__init__.py

@@ -1 +1 @@
-from app.translator.platforms.arcsight.renders.arcsight_cti import ArcsightKeyword
+from app.translator.platforms.arcsight.renders.arcsight_cti import ArcsightKeyword  # noqa: F401

uncoder-core/app/translator/platforms/arcsight/const.py

@@ -0,0 +1,8 @@
+ARCSIGHT_QUERY_DETAILS = {
+    "platform_id": "arcsight",
+    "name": "ArcSight Query",
+    "group_name": "ArcSight",
+    "group_id": "arcsight",
+    "platform_name": "Query",
+    "alt_platform_name": "CEF",
+}

uncoder-core/app/translator/platforms/arcsight/mappings/__init__.py

@@ -0,0 +1,12 @@
+DEFAULT_ARCSIGHT_MAPPING = {
+    "SourceIP": "sourceAddress",
+    "DestinationIP": "destinationAddress",
+    "Domain": "destinationDnsDomain",
+    "URL": "requestUrl",
+    "HashMd5": "fileHash",
+    "HashSha1": "fileHash",
+    "HashSha256": "fileHash",
+    "HashSha512": "fileHash",
+    "Emails": "sender-address",
+    "Files": "winlog.event_data.TargetFilename",
+}

uncoder-core/app/translator/platforms/arcsight/mappings/arcsight_cti.py

@@ -1,3 +1,3 @@
-from app.translator.platforms.athena.parsers.athena import AthenaQueryParser
-from app.translator.platforms.athena.renders.athena import AthenaQueryRender
-from app.translator.platforms.athena.renders.athena_cti import AthenaCTI
+from app.translator.platforms.athena.parsers.athena import AthenaQueryParser  # noqa: F401
+from app.translator.platforms.athena.renders.athena import AthenaQueryRender  # noqa: F401
+from app.translator.platforms.athena.renders.athena_cti import AthenaCTI  # noqa: F401

uncoder-core/app/translator/platforms/arcsight/renders/__init__.py

@@ -0,0 +1,3 @@
+UTF8_PAYLOAD_PATTERN = r"UTF8\(payload\)"
+NUM_VALUE_PATTERN = r"(?P<num_value>\d+(?:\.\d+)*)"
+SINGLE_QUOTES_VALUE_PATTERN = r"""'(?P<s_q_value>(?:[:a-zA-Z\*0-9=+%#\-\/\\,_".$&^@!\(\)\{\}\s]|'')*)'"""

uncoder-core/app/translator/platforms/athena/__init__.py

@@ -0,0 +1,8 @@
+from app.translator.core.escape_manager import EscapeManager
+
+
+class AQLEscapeManager(EscapeManager):
+    ...
+
+
+aql_escape_manager = AQLEscapeManager()

uncoder-core/app/translator/platforms/base/aql/__init__.py

@@ -0,0 +1,122 @@
+"""
+Uncoder IO Community Edition License
+-----------------------------------------------------------------
+Copyright (c) 2024 SOC Prime, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-----------------------------------------------------------------
+"""
+from typing import Union
+
+from app.translator.const import DEFAULT_VALUE_TYPE
+from app.translator.core.custom_types.values import ValueType
+from app.translator.core.render import BaseQueryFieldValue, PlatformQueryRender
+from app.translator.platforms.base.aql.escape_manager import aql_escape_manager
+from app.translator.platforms.base.aql.mapping import AQLLogSourceSignature, AQLMappings, aql_mappings
+
+
+class AQLFieldValue(BaseQueryFieldValue):
+    escape_manager = aql_escape_manager
+
+    def apply_value(self, value: Union[str, int], value_type: str = ValueType.value) -> Union[str, int]:  # noqa: ARG002
+        if isinstance(value, str):
+            value = value.replace("_", "__").replace("%", "%%").replace("\\'", "%").replace("'", '"')
+            if value.endswith("\\\\%"):
+                value = value.replace("\\\\%", "\\%")
+        return value
+
+    def _apply_value(self, value: Union[str, int]) -> Union[str, int]:
+        if isinstance(value, str) and "\\" in value:
+            return value
+        return self.apply_value(value)
+
+    def equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
+        if isinstance(value, list):
+            return f"({self.or_token.join([self.equal_modifier(field=field, value=v) for v in value])})"
+        if field == "UTF8(payload)":
+            return f"UTF8(payload) ILIKE '{self.apply_value(value)}'"
+        if isinstance(value, int):
+            return f'"{field}"={value}'
+
+        return f"\"{field}\"='{self._apply_value(value)}'"
+
+    def less_modifier(self, field: str, value: Union[int, str]) -> str:
+        if isinstance(value, int):
+            return f'"{field}"<{value}'
+        return f"\"{field}\"<'{self._apply_value(value)}'"
+
+    def less_or_equal_modifier(self, field: str, value: Union[int, str]) -> str:
+        if isinstance(value, int):
+            return f'"{field}"<={value}'
+        return f"\"{field}\"<='{self._apply_value(value)}'"
+
+    def greater_modifier(self, field: str, value: Union[int, str]) -> str:
+        if isinstance(value, int):
+            return f'"{field}">{value}'
+        return f"\"{field}\">'{self._apply_value(value)}'"
+
+    def greater_or_equal_modifier(self, field: str, value: Union[int, str]) -> str:
+        if isinstance(value, int):
+            return f'"{field}">={value}'
+        return f"\"{field}\">='{self._apply_value(value)}'"
+
+    def not_equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
+        if isinstance(value, list):
+            return f"({self.or_token.join([self.not_equal_modifier(field=field, value=v) for v in value])})"
+        if isinstance(value, int):
+            return f'"{field}"!={value}'
+        return f"\"{field}\"!='{self._apply_value(value)}'"
+
+    def contains_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
+        if isinstance(value, list):
+            return f"({self.or_token.join(self.contains_modifier(field=field, value=v) for v in value)})"
+        return f"\"{field}\" ILIKE '%{self._apply_value(value)}%'"
+
+    def endswith_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
+        if isinstance(value, list):
+            return f"({self.or_token.join(self.endswith_modifier(field=field, value=v) for v in value)})"
+        return f"\"{field}\" ILIKE '%{self._apply_value(value)}'"
+
+    def startswith_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
+        if isinstance(value, list):
+            return f"({self.or_token.join(self.startswith_modifier(field=field, value=v) for v in value)})"
+        return f"\"{field}\" ILIKE '{self._apply_value(value)}%'"
+
+    def regex_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
+        if isinstance(value, list):
+            return f"({self.or_token.join(self.regex_modifier(field=field, value=v) for v in value)})"
+        return f"\"{field}\" IMATCHES '{value}'"
+
+    def keywords(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
+        if isinstance(value, list):
+            return f"({self.or_token.join(self.keywords(field=field, value=v) for v in value)})"
+        return f"UTF8(payload) ILIKE '%{self.apply_value(value)}%'"
+
+
+class AQLQueryRender(PlatformQueryRender):
+    mappings: AQLMappings = aql_mappings
+
+    or_token = "OR"
+    and_token = "AND"
+    not_token = "NOT"
+
+    field_value_map = AQLFieldValue(or_token=or_token)
+    query_pattern = "{prefix} AND {query} {functions}"
+
+    def generate_prefix(self, log_source_signature: AQLLogSourceSignature) -> str:
+        table = str(log_source_signature)
+        extra_condition = log_source_signature.extra_condition
+        return f"SELECT UTF8(payload) FROM {table} WHERE {extra_condition}"
+
+    def wrap_with_comment(self, value: str) -> str:
+        return f"/* {value} */"