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",
+}