Merge pull request #149 from UncoderIO/gis-7581

Fix empty index field in elastic rule

Author: nazargesyk

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

@@ -20,6 +20,10 @@ def is_suitable(self, *args, **kwargs) -> bool:
     def __str__(self) -> str:
         raise NotImplementedError("Abstract method")
 
+    @property
+    def default_source(self) -> dict:
+        return self._default_source
+
 
 class FieldMapping:
     def __init__(self, generic_field_name: str, platform_field_name: str):

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

@@ -10,8 +10,8 @@ class AQLEscapeManager(EscapeManager):
         ValueType.value: [EscapeDetails(pattern=r"(')", escape_symbols=r"'\1")],
         ValueType.regex_value: [
             EscapeDetails(pattern=r"([$^*+()\[\]{}|.?\-\\])", escape_symbols=r"\\\1"),
-            EscapeDetails(pattern=r"(')", escape_symbols=r"'\1")
-        ]
+            EscapeDetails(pattern=r"(')", escape_symbols=r"'\1"),
+        ],
     }
 
 

uncoder-core/app/translator/platforms/elasticsearch/renders/detection_rule.py

@@ -85,13 +85,14 @@ def finalize_query(
         query: str,
         functions: str,
         meta_info: Optional[MetaInfoContainer] = None,
-        source_mapping: Optional[SourceMapping] = None,  # noqa: ARG002
+        source_mapping: Optional[SourceMapping] = None,
         not_supported_functions: Optional[list] = None,
         *args,  # noqa: ARG002
         **kwargs,  # noqa: ARG002
     ) -> str:
         query = super().finalize_query(prefix=prefix, query=query, functions=functions)
         rule = copy.deepcopy(ELASTICSEARCH_DETECTION_RULE)
+        index = source_mapping.log_source_signature.default_source.get("index") if source_mapping else None
         rule.update(
             {
                 "query": query,
@@ -105,6 +106,7 @@ def finalize_query(
                 "tags": meta_info.tags,
                 "threat": self.__create_mitre_threat(meta_info.mitre_attack),
                 "false_positives": meta_info.false_positives,
+                "index": [index] if index else [],
             }
         )
         rule_str = json.dumps(rule, indent=4, sort_keys=False, ensure_ascii=False)