fix conflicts

Author: dtsprm

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

@@ -116,7 +116,7 @@ def prepare_mapping(self) -> dict[str, SourceMapping]:
         default_mapping = SourceMapping(source_id=DEFAULT_MAPPING_NAME)
         for mapping_dict in self._loader.load_platform_mappings(self._platform_dir):
             log_source_signature = self.prepare_log_source_signature(mapping=mapping_dict)
-            if (source_id := mapping_dict["source"]) == DEFAULT_MAPPING_NAME:
+            if (source_id := mapping_dict.get("source")) == DEFAULT_MAPPING_NAME:
                 default_mapping.log_source_signature = log_source_signature
                 if self.skip_load_default_mappings:
                     continue

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

@@ -189,7 +189,7 @@ def __load_mitre_configs_from_files(self) -> None:
                         technique_id=technique_data["technique_id"],
                         name=technique_data["technique"],
                         url=technique_data["url"],
-                        tactic=technique_data["tactic"],
+                        tactic=technique_data.get("tactic", []),
                     )
                     self.techniques.insert(technique_id, technique)
         except JSONDecodeError:

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

@@ -8,8 +8,6 @@
 from app.translator.core.mapping import DEFAULT_MAPPING_NAME
 from app.translator.core.models.functions.base import ParsedFunctions
 from app.translator.core.models.query_tokens.field import Field
-<<<<<<< HEAD
-=======
 
 
 @dataclass
@@ -46,7 +44,6 @@ def __init__(
         self.trigger_threshold = trigger_threshold
         self.query_frequency = query_frequency
         self.query_period = query_period
->>>>>>> main
 
 
 class MetaInfoContainer:

uncoder-core/app/translator/core/models/query_tokens/field.py

@@ -1,7 +1,4 @@
-<<<<<<< HEAD
-=======
 from abc import ABC, abstractmethod
->>>>>>> main
 from typing import Optional
 
 from app.translator.core.mapping import DEFAULT_MAPPING_NAME, SourceMapping
@@ -41,13 +38,10 @@ def set_generic_names_map(self, source_mappings: list[SourceMapping], default_ma
 class PredefinedField:
     def __init__(self, name: str):
         self.name = name
-<<<<<<< HEAD
-=======
 
 
 class BaseFieldsGetter(ABC):
     @property
     @abstractmethod
     def fields(self) -> list[Field]:
         raise NotImplementedError("Abstract method")
->>>>>>> main

uncoder-core/app/translator/core/models/query_tokens/field_field.py

@@ -1,16 +1,8 @@
-<<<<<<< HEAD
-from app.translator.core.models.query_tokens.field import Alias, Field
-from app.translator.core.models.query_tokens.identifier import Identifier
-
-
-class FieldField:
-=======
 from app.translator.core.models.query_tokens.field import Alias, BaseFieldsGetter, Field
 from app.translator.core.models.query_tokens.identifier import Identifier
 
 
 class FieldField(BaseFieldsGetter):
->>>>>>> main
     def __init__(
         self,
         source_name_left: str,
@@ -24,8 +16,6 @@ def __init__(
         self.operator = operator
         self.field_right = Field(source_name=source_name_right) if not is_alias_right else None
         self.alias_right = Alias(name=source_name_right) if is_alias_right else None
-<<<<<<< HEAD
-=======
 
     @property
     def fields(self) -> list[Field]:
@@ -36,4 +26,3 @@ def fields(self) -> list[Field]:
             fields.append(self.field_right)
 
         return fields
->>>>>>> main

uncoder-core/app/translator/core/models/query_tokens/field_value.py

@@ -1,21 +1,13 @@
 from typing import Union
 
 from app.translator.core.custom_types.tokens import STR_SEARCH_OPERATORS
-<<<<<<< HEAD
-from app.translator.core.models.query_tokens.field import Alias, Field, PredefinedField
-=======
 from app.translator.core.models.query_tokens.field import Alias, BaseFieldsGetter, Field, PredefinedField
->>>>>>> main
 from app.translator.core.models.query_tokens.identifier import Identifier
 from app.translator.core.models.query_tokens.value import Value
 from app.translator.core.str_value_manager import StrValue
 
 
-<<<<<<< HEAD
-class FieldValue(Value):
-=======
 class FieldValue(BaseFieldsGetter, Value):
->>>>>>> main
     def __init__(
         self,
         source_name: str,
@@ -41,10 +33,7 @@ def __repr__(self):
             return f"{self.predefined_field.name} {self.operator.token_type} {self.values}"
 
         return f"{self.field.source_name} {self.operator.token_type} {self.values}"
-<<<<<<< HEAD
-=======
 
     @property
     def fields(self) -> list[Field]:
         return [self.field] if self.field else []
->>>>>>> main

uncoder-core/app/translator/core/models/query_tokens/function_value.py

@@ -2,28 +2,18 @@
 
 from app.translator.core.custom_types.tokens import STR_SEARCH_OPERATORS
 from app.translator.core.models.functions.base import Function
-<<<<<<< HEAD
-=======
 from app.translator.core.models.query_tokens.field import BaseFieldsGetter, Field
->>>>>>> main
 from app.translator.core.models.query_tokens.identifier import Identifier
 from app.translator.core.models.query_tokens.value import Value
 from app.translator.core.str_value_manager import StrValue
 
 
-<<<<<<< HEAD
-class FunctionValue(Value):
-=======
 class FunctionValue(BaseFieldsGetter, Value):
->>>>>>> main
     def __init__(self, function: Function, operator: Identifier, value: Union[int, str, StrValue, list, tuple]):
         super().__init__(value, cast_to_int=operator.token_type not in STR_SEARCH_OPERATORS)
         self.function = function
         self.operator = operator
-<<<<<<< HEAD
-=======
 
     @property
     def fields(self) -> list[Field]:
         return self.function.fields
->>>>>>> main

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

@@ -48,44 +48,11 @@ class AQLMappings(BasePlatformMappings):
 
     def prepare_log_source_signature(self, mapping: dict) -> AQLLogSourceSignature:
         log_source = mapping.get("log_source", {})
-        default_log_source = mapping["default_log_source"]
+        default_log_source = mapping.get("default_log_source")
         return AQLLogSourceSignature(
             device_types=log_source.get("devicetype"),
             categories=log_source.get("category"),
             qids=log_source.get("qid"),
             qid_event_categories=log_source.get("qideventcategory"),
             default_source=default_log_source,
         )
-<<<<<<< HEAD
-
-    def get_suitable_source_mappings(
-        self,
-        field_names: list[str],
-        devicetype: Optional[list[int]] = None,
-        category: Optional[list[int]] = None,
-        qid: Optional[list[int]] = None,
-        qideventcategory: Optional[list[int]] = None,
-    ) -> list[SourceMapping]:
-        suitable_source_mappings = []
-        for source_mapping in self._source_mappings.values():
-            if source_mapping.source_id == DEFAULT_MAPPING_NAME:
-                continue
-
-            log_source_signature: AQLLogSourceSignature = source_mapping.log_source_signature
-            if log_source_signature.is_suitable(devicetype, category, qid, qideventcategory):  # noqa: SIM102
-                if source_mapping.fields_mapping.is_suitable(field_names):
-                    suitable_source_mappings.append(source_mapping)
-
-        if not suitable_source_mappings:
-            for source_mapping in self._source_mappings.values():
-                if source_mapping.source_id == DEFAULT_MAPPING_NAME:
-                    continue
-                if source_mapping.fields_mapping.is_suitable(field_names):
-                    suitable_source_mappings.append(source_mapping)
-
-        if not suitable_source_mappings:
-            suitable_source_mappings = [self._source_mappings[DEFAULT_MAPPING_NAME]]
-
-        return suitable_source_mappings
-=======
->>>>>>> main

uncoder-core/app/translator/platforms/base/spl/renders/spl.py

@@ -20,15 +20,6 @@
 from typing import Union
 
 from app.translator.const import DEFAULT_VALUE_TYPE
-<<<<<<< HEAD
-from app.translator.core.exceptions.render import UnsupportedRenderMethod
-from app.translator.core.render import BaseFieldValueRender, PlatformQueryRender
-from app.translator.platforms.base.spl.escape_manager import spl_escape_manager
-
-
-class SplFieldValueRender(BaseFieldValueRender):
-    escape_manager = spl_escape_manager
-=======
 from app.translator.core.custom_types.values import ValueType
 from app.translator.core.render import BaseFieldValueRender, PlatformQueryRender
 from app.translator.core.str_value_manager import StrValue
@@ -47,7 +38,6 @@ def _pre_process_value(
     ) -> Union[int, str]:
         value = super()._pre_process_value(field, value, value_type=value_type, wrap_str=wrap_str)
         return self._wrap_str_value(str(value)) if not isinstance(value, str) else value
->>>>>>> main
 
     def equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):

uncoder-core/app/translator/platforms/base/spl/str_value_manager.py

@@ -0,0 +1,55 @@
+"""
+Uncoder IO Community Edition License
+-----------------------------------------------------------------
+Copyright (c) 2023 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 ClassVar
+
+from app.translator.core.str_value_manager import BaseSpecSymbol, StrValue, StrValueManager, UnboundLenWildCard
+from app.translator.platforms.base.spl.escape_manager import spl_escape_manager
+
+
+class SplStrValueManager(StrValueManager):
+    escape_manager = spl_escape_manager
+    str_spec_symbols_map: ClassVar[dict[str, type[BaseSpecSymbol]]] = {"*": UnboundLenWildCard}
+
+    def from_str_to_container(self, value: str) -> StrValue:
+        split = []
+        prev_char = None
+        for char in value:
+            if char == "\\":
+                if prev_char == "\\":
+                    split.append("\\")
+                    prev_char = None
+                    continue
+            elif char in self.str_spec_symbols_map:
+                if prev_char == "\\":
+                    split.append(char)
+                else:
+                    split.append(self.str_spec_symbols_map[char]())
+            elif char in ('"', "=", "|", "<", ">"):
+                split.append(char)
+            else:
+                if prev_char == "\\":
+                    split.append(prev_char)
+                split.append(char)
+
+            prev_char = char
+
+        return StrValue(self.escape_manager.remove_escape(value), self._concat(split))
+
+
+spl_str_value_manager = SplStrValueManager()