Improve roota render + lucene mapping

Author: saltar-ua

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

@@ -36,8 +36,13 @@ def get_suitable_source_mappings(
             if index and log_source_signature.is_suitable(index=index):
                 if source_mapping.fields_mapping.is_suitable(field_names):
                     suitable_source_mappings.append(source_mapping)
-            elif 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]]

uncoder-core/app/translator/platforms/roota/renders/roota.py

@@ -15,6 +15,7 @@
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 -----------------------------------------------------------------
 """
+import copy
 import math
 from datetime import timedelta
 from typing import Optional
@@ -73,6 +74,15 @@ def __render_timeframe(timeframe: timedelta) -> str:
             timeframe_unit = "s"
         return f"{timeframe_value}{timeframe_unit}"
 
+    @staticmethod
+    def __normalize_log_source(log_source: dict) -> dict:
+        prepared_log_source = {}
+        for log_source_key, value in log_source.items():
+            if isinstance(value, list):
+                value = value[0]
+            prepared_log_source[log_source_key] = value.lower()
+        return prepared_log_source
+
     def __get_data_for_roota_render(
         self, raw_query_container: RawQueryContainer, tokenized_query_container: TokenizedQueryContainer
     ) -> tuple:
@@ -90,14 +100,15 @@ def __get_data_for_roota_render(
             return_only_first_query_ctx_var.set(prev_state_return_only_first_query_ctx_var)
             wrap_query_with_meta_info_ctx_var.set(prev_state_wrap_query_with_meta_info_ctx_var)
 
-            return rule_query, rule_query_language, tokenized_query_container.meta_info.parsed_logsources
+            return (rule_query, rule_query_language,
+                    self.__normalize_log_source(log_source=tokenized_query_container.meta_info.parsed_logsources))
         rule_query_language = raw_query_container.language.replace("rule", "query")
         rule_query = raw_query_container.query
         for source_mapping_id in tokenized_query_container.meta_info.source_mapping_ids:
             if source_mapping_id == "default":
                 continue
             if logsources := self.__get_logsources_by_source_mapping_id(source_mapping_id=source_mapping_id):
-                return rule_query, rule_query_language, logsources
+                return rule_query, rule_query_language, self.__normalize_log_source(log_source=logsources)
         return rule_query, rule_query_language, {}
 
     @staticmethod
@@ -114,7 +125,7 @@ def generate(
             raw_query_container=raw_query_container, tokenized_query_container=tokenized_query_container
         )
 
-        rule = ROOTA_RULE_TEMPLATE.copy()
+        rule = copy.deepcopy(ROOTA_RULE_TEMPLATE)
         rule["name"] = tokenized_query_container.meta_info.title or _AUTOGENERATED_TEMPLATE
         rule["details"] = tokenized_query_container.meta_info.description or rule["details"]
         rule["author"] = tokenized_query_container.meta_info.author or rule["author"]
@@ -139,9 +150,6 @@ def generate(
             rule["correlation"]["timeframe"] = self.__render_timeframe(tokenized_query_container.meta_info.timeframe)
 
         if rule_logsources:
-            for logsource_str, value in rule_logsources.items():
-                if isinstance(value, list):
-                    value = value[0]
-                rule["logsource"][logsource_str] = value.lower()
+           rule["logsources"] = rule_logsources
 
         return yaml.dump(rule, Dumper=IndentedListDumper, default_flow_style=False, sort_keys=False, indent=4)