fix

Author: tarnopolskyi

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

@@ -69,14 +69,13 @@ def __init__(
         self.raw_mitre_attack = raw_mitre_attack or []
         self.status = status or "stable"
         self.false_positives = false_positives or []
-        self.source_mapping_ids = source_mapping_ids or [DEFAULT_MAPPING_NAME]
+        self.source_mapping_ids = sorted(source_mapping_ids) if source_mapping_ids else [DEFAULT_MAPPING_NAME]
         self.parsed_logsources = parsed_logsources or {}
         self.timeframe = timeframe
 
     @property
-    def author_str(self) -> Optional[str]:
-        if self.author:
-            return ", ".join(self.author)
+    def author_str(self) -> str:
+        return ", ".join(self.author)
 
 
 @dataclass

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

@@ -33,14 +33,10 @@ def parse_raw_query(self, text: str, language: str) -> RawQueryContainer:
         rule = self.load_rule(text=text)
         parsed_description = parse_rule_description_str(rule.get("description", ""))
 
-        mitre_attack = None
-        if rule_mitre_attack_data := rule.get("threat"):
-            mitre_attack = self.mitre_config.get_mitre_info(
-                tactics=[
-                    threat_data["tactic"]["name"].replace(" ", "_").lower() for threat_data in rule_mitre_attack_data
-                ],
-                techniques=[threat_data["technique"][0]["id"].lower() for threat_data in rule_mitre_attack_data],
-            )
+        mitre_attack = self.mitre_config.get_mitre_info(
+            tactics=[threat_data["tactic"]["name"].replace(" ", "_").lower() for threat_data in rule.get("threat", [])],
+            techniques=[threat_data["technique"][0]["id"].lower() for threat_data in rule.get("threat", [])],
+        )
 
         return RawQueryContainer(
             query=rule["query"],

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

@@ -22,8 +22,9 @@
 from typing import Optional
 
 from app.translator.core.mapping import SourceMapping
+from app.translator.core.mitre import MitreConfig
 from app.translator.core.models.platform_details import PlatformDetails
-from app.translator.core.models.query_container import MetaInfoContainer
+from app.translator.core.models.query_container import MetaInfoContainer, MitreInfoContainer
 from app.translator.managers import render_manager
 from app.translator.platforms.base.lucene.mapping import LuceneMappings
 from app.translator.platforms.elasticsearch.const import XPACK_WATCHER_RULE, xpack_watcher_details
@@ -47,6 +48,24 @@ class XPackWatcherRuleRender(ElasticSearchQueryRender):
     mappings: LuceneMappings = xpack_watcher_mappings
     or_token = "OR"
     field_value_render = XpackWatcherRuleFieldValue(or_token=or_token)
+    mitre: MitreConfig = MitreConfig()
+
+    def __create_mitre_threat(self, mitre_attack: MitreInfoContainer) -> dict:
+        result = {"tactics": [], "techniques": []}
+
+        for tactic in mitre_attack.tactics:
+            result["tactics"].append({"external_id": tactic.external_id, "url": tactic.url, "tactic": tactic.name})
+        for technique in mitre_attack.techniques:
+            result["techniques"].append(
+                {
+                    "technique_id": technique.technique_id,
+                    "technique": technique.name,
+                    "url": technique.url,
+                    "tactic": technique.tactic,
+                }
+            )
+
+        return result if result["tactics"] or result["techniques"] else {}
 
     def finalize_query(
         self,
@@ -76,7 +95,7 @@ def finalize_query(
                     license_=meta_info.license,
                     mitre_attack=mitre_attack,
                 ),
-                "tags": meta_info.tags,
+                "tags": self.__create_mitre_threat(mitre_attack=meta_info.mitre_attack),
             }
         )
         rule["input"]["search"]["request"]["body"]["query"]["bool"]["must"][0]["query_string"]["query"] = query

uncoder-core/app/translator/platforms/splunk/parsers/splunk_alert.py

@@ -18,6 +18,7 @@
 
 import re
 
+from app.translator.core.custom_types.meta_info import SeverityType
 from app.translator.core.models.platform_details import PlatformDetails
 from app.translator.core.models.query_container import MetaInfoContainer, RawQueryContainer
 from app.translator.managers import parser_manager
@@ -36,9 +37,15 @@ def parse_raw_query(self, text: str, language: str) -> RawQueryContainer:
         rule_name: str = ""
         severity: str = ""
         raw_mitre_attack: list[str] = []
-        if severity_match := re.search(r"action\.risk\.param\._risk_score\s*=\s*(\d+)", text):
-            level_map = {"0": "informational", "25": "low", "50": "medium", "75": "high", "100": "critical"}
+        if severity_match := re.search(r"alert\.severity\s*=\s*(\d+)", text):
+            level_map = {
+                "1": SeverityType.low,
+                "2": SeverityType.medium,
+                "3": SeverityType.high,
+                "4": SeverityType.critical,
+            }
             severity = level_map.get(str(severity_match.group(1)), "low")
+
         if mitre_attack_match := re.search(r'"mitre_attack":\s*\[(.*?)\]', text):
             raw_mitre_attack = [attack.strip().strip('"').lower() for attack in mitre_attack_match.group(1).split(",")]
 

uncoder-core/app/translator/platforms/splunk/renders/splunk_alert.py

@@ -22,7 +22,7 @@
 from app.translator.core.custom_types.meta_info import SeverityType
 from app.translator.core.mapping import SourceMapping
 from app.translator.core.models.platform_details import PlatformDetails
-from app.translator.core.models.query_container import MetaInfoContainer
+from app.translator.core.models.query_container import MetaInfoContainer, MitreInfoContainer
 from app.translator.managers import render_manager
 from app.translator.platforms.splunk.const import DEFAULT_SPLUNK_ALERT, splunk_alert_details
 from app.translator.platforms.splunk.mapping import SplunkMappings, splunk_alert_mappings
@@ -46,13 +46,15 @@ class SplunkAlertRender(SplunkQueryRender):
     field_value_render = SplunkAlertFieldValueRender(or_token=or_token)
 
     @staticmethod
-    def __create_mitre_threat(meta_info: MetaInfoContainer) -> dict:
-        techniques = {"mitre_attack": []}
+    def __create_mitre_threat(mitre_attack: MitreInfoContainer) -> dict:
+        mitre_attack_render = {"mitre_attack": []}
 
-        for technique in meta_info.mitre_attack.techniques:
-            techniques["mitre_attack"].append(technique.technique_id)
-        techniques["mitre_attack"].sort()
-        return techniques
+        for technique in mitre_attack.techniques:
+            mitre_attack_render["mitre_attack"].append(technique.technique_id)
+        for tactic in mitre_attack.tactics:
+            mitre_attack_render["mitre_attack"].append(tactic.name)
+        mitre_attack_render["mitre_attack"].sort()
+        return mitre_attack_render
 
     def finalize_query(
         self,
@@ -77,7 +79,7 @@ def finalize_query(
             rule_id=meta_info.id,
         )
         rule = rule.replace("<description_place_holder>", rule_description)
-        mitre_techniques = self.__create_mitre_threat(meta_info=meta_info)
+        mitre_techniques = self.__create_mitre_threat(mitre_attack=meta_info.mitre_attack)
         if mitre_techniques:
             mitre_str = f"action.correlationsearch.annotations = {mitre_techniques})"
             rule = rule.replace("<annotations_place_holder>", mitre_str)