Merge pull request #23 from UncoderIO/added-mittre-attack-tags-parsing

added mitre-attack parsing + mitre-attack render to platforms

Author: alex-bredikhin

siem-converter/app/converter/core/mitre.py

@@ -0,0 +1,108 @@
+import json
+import os
+import urllib.request
+import ssl
+from urllib.error import HTTPError
+
+from app.converter.tools.singleton_meta import SingletonMeta
+from const import ROOT_PROJECT_PATH
+
+
+class MitreConfig(metaclass=SingletonMeta):
+    config_url: str = 'https://raw.githubusercontent.com/mitre/cti/master/enterprise-attack/enterprise-attack.json'
+    mitre_source_types: tuple = ('mitre-attack', )
+    tactics: dict = {}
+    techniques: dict = {}
+
+    @staticmethod
+    def __revoked_or_deprecated(entry: dict) -> bool:
+        if entry.get("revoked") or entry.get("x_mitre_deprecated"):
+            return True
+        return False
+
+    def __get_mitre_json(self) -> dict:
+        ctx = ssl.create_default_context()
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+
+        try:
+            with urllib.request.urlopen(self.config_url, context=ctx) as cti_json:
+                return json.loads(cti_json.read().decode())
+        except HTTPError:
+            return {}
+    def update_mitre_config(self) -> None:
+        if not (mitre_json := self.__get_mitre_json()):
+            self.__load_mitre_configs_from_files()
+            return
+
+        tactic_map = {}
+        technique_map = {}
+
+        # Map the tatics
+        for entry in mitre_json["objects"]:
+            if not entry["type"] == "x-mitre-tactic" or self.__revoked_or_deprecated(entry):
+                continue
+            for ref in entry["external_references"]:
+                if ref["source_name"] == 'mitre-attack':
+                    tactic_map[entry["x_mitre_shortname"]] = entry["name"]
+                    self.tactics[entry["name"].replace(' ', '_').lower()] = {
+                        "external_id": ref["external_id"],
+                        "url": ref["url"],
+                        "tactic": entry["name"]
+                    }
+                    break
+
+        # Map the techniques
+        for entry in mitre_json["objects"]:
+            if not entry["type"] == "attack-pattern" or self.__revoked_or_deprecated(entry):
+                continue
+            if entry.get("x_mitre_is_subtechnique"):
+                continue
+            for ref in entry["external_references"]:
+                if ref["source_name"] in self.mitre_source_types:
+                    technique_map[ref["external_id"]] = entry["name"]
+                    sub_tactics = []
+                    # Get Mitre Tactics (Kill-Chains)
+                    for tactic in entry["kill_chain_phases"]:
+                        if tactic["kill_chain_name"] in self.mitre_source_types:
+                            # Map the short phase_name to tactic name
+                            sub_tactics.append(tactic_map[tactic["phase_name"]])
+                    self.techniques[ref["external_id"].lower()] = {
+                        "technique_id": ref["external_id"],
+                        "technique": entry["name"],
+                        "url": ref["url"],
+                        "tactic": sub_tactics
+                    }
+                    break
+
+        ## Map the sub-techniques
+        for entry in mitre_json["objects"]:
+            if not entry["type"] == "attack-pattern" or self.__revoked_or_deprecated(entry):
+                continue
+            if entry.get("x_mitre_is_subtechnique"):
+                for ref in entry["external_references"]:
+                    if ref["source_name"] in self.mitre_source_types:
+                        sub_technique_id = ref["external_id"]
+                        sub_technique_name = entry["name"]
+                        parent_technique_name = technique_map[sub_technique_id.split(".")[0]]
+                        sub_technique_name = "{} : {}".format(parent_technique_name, sub_technique_name)
+                        self.techniques[ref["external_id"].lower()] = {
+                            "technique_id": ref["external_id"],
+                            "technique": sub_technique_name,
+                            "url": ref["url"],
+                        }
+                        break
+
+    def __load_mitre_configs_from_files(self) -> None:
+        with open(os.path.join(ROOT_PROJECT_PATH, 'app/dictionaries/tactics.json'), 'r') as file:
+            self.tactics = json.load(file)
+
+        with open(os.path.join(ROOT_PROJECT_PATH, 'app/dictionaries/techniques.json'), 'r') as file:
+            self.techniques = json.load(file)
+
+    def get_tactic(self, tactic: str) -> dict:
+        tactic = tactic.replace('.', '_')
+        return self.tactics.get(tactic, {})
+
+    def get_technique(self, technique_id: str) -> dict:
+        return self.techniques.get(technique_id, {})

siem-converter/app/converter/core/mixins/rule.py

@@ -1,8 +1,10 @@
 import json
+from typing import List
 
 import yaml
 
 from app.converter.core.exceptions.core import InvalidYamlStructure, InvalidJSONStructure
+from app.converter.core.mitre import MitreConfig
 
 
 class JsonRuleMixin:
@@ -15,9 +17,28 @@ def load_rule(self, text):
 
 
 class YamlRuleMixin:
+    mitre_config: MitreConfig = MitreConfig()
 
     def load_rule(self, text):
         try:
             return yaml.safe_load(text)
         except yaml.YAMLError as err:
             raise InvalidYamlStructure(error=str(err))
+
+    def parse_mitre_attack(self, tags: List[str]) -> dict[str, list]:
+        result = {
+            'tactics': [],
+            'techniques': []
+        }
+        for tag in tags:
+            tag = tag.lower()
+            if tag.startswith('attack.'):
+                tag = tag[7::]
+            if tag.startswith('t'):
+                if technique := self.mitre_config.get_technique(tag):
+                    result['techniques'].append(technique)
+            else:
+                if tactic := self.mitre_config.get_tactic(tag):
+                    result['tactics'].append(tactic)
+
+        return result

siem-converter/app/converter/core/models/parser_output.py

@@ -17,8 +17,8 @@ def __init__(self, *,
                  license_: str = None,
                  severity: str = None,
                  references: List[str] = None,
-                 tags: List[str] = None,
-                 mitre_attack: List[str] = None,
+                 tags: list[str] = None,
+                 mitre_attack: dict[str, list] = None,
                  status: str = None,
                  false_positives: List[str] = None,
                  source_mapping_ids: List[str] = None

siem-converter/app/converter/platforms/chronicle/const.py

@@ -7,8 +7,10 @@
     version = "0.01"
     rule_id = "<rule_id_place_holder>"
     status = "<status_place_holder>"
-    severity = "<severity_place_holder>"
+    tags = "<tags_place_holder>"
     falsepositives = "<falsepositives_place_holder>"
+    severity = "<severity_place_holder>"
+    created = "<created_place_holder>"
 
   events:
     <query_placeholder>

siem-converter/app/converter/platforms/chronicle/renders/chronicle_rule.py

@@ -96,4 +96,6 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
         rule = rule.replace("<severity_place_holder>", meta_info.severity)
         rule = rule.replace("<status_place_holder>", meta_info.status)
         rule = rule.replace("<falsepositives_place_holder>", ', '.join(meta_info.false_positives))
+        rule = rule.replace("<tags_place_holder>", ", ".join(meta_info.tags))
+        rule = rule.replace("<created_place_holder>", str(meta_info.date))
         return rule

siem-converter/app/converter/platforms/elasticsearch/renders/detection_rule.py

@@ -19,6 +19,7 @@
 
 import copy
 import json
+from typing import Union
 
 from app.converter.platforms.elasticsearch.const import ELASTICSEARCH_DETECTION_RULE, elasticsearch_rule_details
 from app.converter.platforms.elasticsearch.mapping import ElasticSearchMappings, elasticsearch_mappings
@@ -27,6 +28,7 @@
 from app.converter.core.models.platform_details import PlatformDetails
 from app.converter.core.models.parser_output import MetaInfoContainer
 from app.converter.tools.utils import concatenate_str, get_mitre_attack_str
+from app.converter.core.mitre import MitreConfig
 
 
 class ElasticSearchRuleFieldValue(ElasticSearchFieldValue):
@@ -36,6 +38,7 @@ class ElasticSearchRuleFieldValue(ElasticSearchFieldValue):
 class ElasticSearchRuleRender(ElasticSearchQueryRender):
     details: PlatformDetails = elasticsearch_rule_details
     mappings: ElasticSearchMappings = elasticsearch_mappings
+    mitre: MitreConfig = MitreConfig()
 
     or_token = "OR"
     and_token = "AND"
@@ -44,6 +47,46 @@ class ElasticSearchRuleRender(ElasticSearchQueryRender):
     field_value_map = ElasticSearchRuleFieldValue(or_token=or_token)
     query_pattern = "{prefix} {query} {functions}"
 
+    def __create_mitre_threat(self, mitre_attack: dict) -> Union[list, list[dict]]:
+        if not mitre_attack.get('techniques'):
+            return []
+        threat = []
+
+        if not mitre_attack.get('tactics'):
+            for technique in mitre_attack.get('techniques'):
+                technique_name = technique['technique']
+                if '.' in technique_name:
+                    technique_name = technique_name[:technique_name.index('.')]
+                threat.append(technique_name)
+            return threat
+
+        for tactic in mitre_attack['tactics']:
+            tactic_render = {
+                'id': tactic['external_id'],
+                'name': tactic['tactic'],
+                'reference': tactic['url']
+            }
+            sub_threat = {
+                'tactic': tactic_render,
+                'framework': 'MITRE ATT&CK',
+                'technique': []
+            }
+            for technique in mitre_attack['techniques']:
+                technique_id = technique['technique_id'].lower()
+                if '.' in technique_id:
+                    technique_id = technique_id[:technique['technique_id'].index('.')]
+                main_technique = self.mitre.get_technique(technique_id)
+                if tactic['tactic'] in main_technique['tactic']:
+                    sub_threat['technique'].append({
+                        "id": main_technique['technique_id'],
+                        "name": main_technique['technique'],
+                        "reference": main_technique['url']
+                    })
+            if len(sub_threat['technique']) > 0:
+                threat.append(sub_threat)
+
+        return threat
+
     def finalize_query(self, prefix: str, query: str, functions: str, meta_info: MetaInfoContainer,
                        source_mapping: SourceMapping = None, not_supported_functions: list = None):
         query = super().finalize_query(prefix=prefix, query=query, functions=functions, meta_info=meta_info)
@@ -61,7 +104,8 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
             "severity": meta_info.severity,
             "references": meta_info.references,
             "license": meta_info.license,
-            "tags": meta_info.mitre_attack,
+            "tags": meta_info.tags,
+            "threat": self.__create_mitre_threat(meta_info.mitre_attack),
             "false_positives": meta_info.false_positives
         })
         rule_str = json.dumps(rule, indent=4, sort_keys=False, ensure_ascii=False)

siem-converter/app/converter/platforms/elasticsearch/renders/elast_alert.py

@@ -52,8 +52,7 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
             "<description_place_holder>",
             get_rule_description_str(
                 description=meta_info.description,
-                license=meta_info.license,
-                mitre_attack=meta_info.mitre_attack
+                license=meta_info.license
             )
         )
         rule = rule.replace("<title_place_holder>", meta_info.title)

siem-converter/app/converter/platforms/elasticsearch/renders/kibana.py

@@ -53,8 +53,7 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
             author=meta_info.author,
             rule_id=meta_info.id,
             license=meta_info.license,
-            references=meta_info.references,
-            mitre_attack=meta_info.mitre_attack
+            references=meta_info.references
         )
         rule_str = json.dumps(rule, indent=4, sort_keys=False)
         if not_supported_functions:

siem-converter/app/converter/platforms/logscale/renders/logscale_alert.py

@@ -25,7 +25,7 @@
 from app.converter.core.mapping import SourceMapping
 from app.converter.core.models.platform_details import PlatformDetails
 from app.converter.core.models.parser_output import MetaInfoContainer
-from app.converter.tools.utils import get_rule_description_str
+from app.converter.tools.utils import get_rule_description_str, get_mitre_attack_str
 
 
 _AUTOGENERATED_TITLE = "Autogenerated Falcon LogScale Alert"
@@ -45,10 +45,14 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
         rule = copy.deepcopy(DEFAULT_LOGSCALE_ALERT)
         rule['query']['queryString'] = query
         rule['name'] = meta_info.title or _AUTOGENERATED_TITLE
+        mitre_attack = []
+        if meta_info.mitre_attack:
+            mitre_attack = [f"ATTACK.{i['tactic']}" for i in meta_info.mitre_attack.get('tactics', [])]
+            mitre_attack.extend([f"ATTACK.{i['technique_id']}" for i in meta_info.mitre_attack.get('techniques', [])])
         rule['description'] = get_rule_description_str(
             description=meta_info.description,
             license=meta_info.license,
-            mitre_attack=meta_info.mitre_attack,
+            mitre_attack=mitre_attack,
             author=meta_info.author
         )
 

siem-converter/app/converter/platforms/microsoft/renders/microsoft_sentinel_rule.py

@@ -40,6 +40,18 @@ class MicrosoftSentinelRuleRender(MicrosoftSentinelQueryRender):
     or_token = "or"
     field_value_map = MicrosoftSentinelRuleFieldValue(or_token=or_token)
 
+    def __create_mitre_threat(self, meta_info: MetaInfoContainer) -> tuple[list, list]:
+        tactics = []
+        techniques = []
+
+        for tactic in meta_info.mitre_attack.get('tactics'):
+            tactics.append(tactic['tactic'])
+
+        for technique in meta_info.mitre_attack.get('techniques'):
+            techniques.append(technique['technique_id'])
+
+        return tactics, techniques
+
     def finalize_query(self, prefix: str, query: str, functions: str, meta_info: MetaInfoContainer,
                        source_mapping: SourceMapping = None, not_supported_functions: list = None):
         query = super().finalize_query(prefix=prefix, query=query, functions=functions, meta_info=meta_info)
@@ -52,7 +64,9 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
             license=meta_info.license
         )
         rule["severity"] = meta_info.severity
-        rule["techniques"] = [el.upper() for el in meta_info.mitre_attack]
+        mitre_tactics, mitre_techniques = self.__create_mitre_threat(meta_info=meta_info)
+        rule['tactics'] = mitre_tactics
+        rule['techniques'] = mitre_techniques
         json_rule = json.dumps(rule, indent=4, sort_keys=False)
         if not_supported_functions:
             rendered_not_supported = self.render_not_supported_functions(not_supported_functions)

siem-converter/app/converter/platforms/roota/parsers/roota.py

@@ -15,6 +15,7 @@
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 -----------------------------------------------------------------
 """
+import re
 
 from app.converter.core.exceptions.core import UnsupportedRootAParser, RootARuleValidationException
 from app.converter.core.mixins.rule import YamlRuleMixin
@@ -27,16 +28,19 @@ class RootAParser(YamlRuleMixin):
     parsers = parser_manager
     mandatory_fields = {"name", "details", "author", "severity", "mitre-attack", "detection", "references", "license"}
 
-    @staticmethod
-    def __update_meta_info(meta_info: MetaInfoContainer, rule: dict) -> MetaInfoContainer:
+    def __update_meta_info(self, meta_info: MetaInfoContainer, rule: dict) -> MetaInfoContainer:
         mitre_attack = rule.get("mitre-attack") or []
-        mitre_attack = [i.strip("") for i in mitre_attack.split(",")] if isinstance(mitre_attack, str) else mitre_attack
+        mitre_tags = [i.strip("") for i in mitre_attack.split(",")] if isinstance(mitre_attack, str) else mitre_attack
+        mitre_attack = self.parse_mitre_attack(mitre_tags)
+        rule_tags = rule.get('tags', [])
+        rule_tags += mitre_tags
+
         meta_info.title = rule.get("name")
         meta_info.description = rule.get("details")
         meta_info.id = rule.get("uuid", meta_info.id)
         meta_info.references = rule.get("references")
         meta_info.license = rule.get("license", meta_info.license)
-        meta_info.tags = rule.get("tags", meta_info.tags)
+        meta_info.tags = rule_tags or meta_info.tags
         meta_info.mitre_attack = mitre_attack
         meta_info.date = rule.get("date", meta_info.date)
         meta_info.author = rule.get("author", meta_info.author)

siem-converter/app/converter/platforms/sigma/parsers/sigma.py

@@ -38,15 +38,6 @@ class SigmaParser(YamlRuleMixin):
     mappings: SigmaMappings = sigma_mappings
     mandatory_fields = {"title", "description", "references", "logsource", "detection"}
 
-    @staticmethod
-    def __parse_mitre_attack(tags: List[str]) -> List[str]:
-        result = []
-        for tag in tags:
-            if search := re.search(r"[tT]\d{4}(?:\.\d{3})?", tag):
-                result.append(search.group())
-
-        return result
-
     @staticmethod
     def __parse_false_positives(false_positives: Union[str, List[str], None]) -> list:
         if isinstance(false_positives, str):
@@ -62,9 +53,10 @@ def _get_meta_info(self, rule: dict, source_mapping_ids: List[str]) -> MetaInfoC
             date=rule.get("date"),
             references=rule.get("references", []),
             license_=rule.get("license"),
-            mitre_attack=self.__parse_mitre_attack(rule.get("tags", [])),
+            mitre_attack=self.parse_mitre_attack(rule.get("tags", [])),
             severity=rule.get("level"),
             status=rule.get("status"),
+            tags=rule.get("tags"),
             false_positives=self.__parse_false_positives(rule.get("falsepositives")),
             source_mapping_ids=source_mapping_ids
         )