Merge pull request #182 from UncoderIO/gis-roota-render

Gis roota render

Author: tarnopolskyi

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

@@ -19,9 +19,14 @@ class LogSourceSignature(ABC):
     wildcard_symbol = "*"
 
     @abstractmethod
-    def is_suitable(self, *args, **kwargs) -> bool:
+    def is_suitable(self, **kwargs) -> bool:
         raise NotImplementedError("Abstract method")
 
+    @staticmethod
+    def _check_conditions(conditions: list[Union[bool, None]]) -> bool:
+        conditions = [condition for condition in conditions if condition is not None]
+        return bool(conditions) and all(conditions)
+
     @abstractmethod
     def __str__(self) -> str:
         raise NotImplementedError("Abstract method")
@@ -147,9 +152,23 @@ def prepare_fields_mapping(field_mapping: dict) -> FieldsMapping:
     def prepare_log_source_signature(self, mapping: dict) -> LogSourceSignature:
         raise NotImplementedError("Abstract method")
 
-    @abstractmethod
-    def get_suitable_source_mappings(self, *args, **kwargs) -> list[SourceMapping]:
-        raise NotImplementedError("Abstract method")
+    def get_suitable_source_mappings(
+        self, field_names: list[str], log_sources: dict[str, list[Union[int, str]]]
+    ) -> list[SourceMapping]:
+        by_log_sources_and_fields = []
+        by_fields = []
+        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):
+                by_fields.append(source_mapping)
+
+                log_source_signature: LogSourceSignature = source_mapping.log_source_signature
+                if log_source_signature and log_source_signature.is_suitable(**log_sources):
+                    by_log_sources_and_fields.append(source_mapping)
+
+        return by_log_sources_and_fields or by_fields or [self._source_mappings[DEFAULT_MAPPING_NAME]]
 
     def get_source_mapping(self, source_id: str) -> Optional[SourceMapping]:
         return self._source_mappings.get(source_id)

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

@@ -3,8 +3,10 @@
 import ssl
 import urllib.request
 from json import JSONDecodeError
+from typing import Optional
 from urllib.error import HTTPError
 
+from app.translator.core.models.query_container import MitreInfoContainer, MitreTacticContainer, MitreTechniqueContainer
 from app.translator.tools.singleton_meta import SingletonMeta
 from const import ROOT_PROJECT_PATH
 
@@ -116,9 +118,34 @@ def __load_mitre_configs_from_files(self) -> None:
         except JSONDecodeError:
             self.techniques = {}
 
-    def get_tactic(self, tactic: str) -> dict:
+    def get_tactic(self, tactic: str) -> Optional[MitreTacticContainer]:
         tactic = tactic.replace(".", "_")
-        return self.tactics.get(tactic, {})
-
-    def get_technique(self, technique_id: str) -> dict:
-        return self.techniques.get(technique_id, {})
+        if tactic_found := self.tactics.get(tactic):
+            return MitreTacticContainer(
+                external_id=tactic_found["external_id"], url=tactic_found["url"], name=tactic_found["tactic"]
+            )
+
+    def get_technique(self, technique_id: str) -> Optional[MitreTechniqueContainer]:
+        if technique_found := self.techniques.get(technique_id):
+            return MitreTechniqueContainer(
+                technique_id=technique_found["technique_id"],
+                name=technique_found["technique"],
+                url=technique_found["url"],
+                tactic=technique_found["tactic"],
+            )
+
+    def get_mitre_info(
+        self, tactics: Optional[list[str]] = None, techniques: Optional[list[str]] = None
+    ) -> MitreInfoContainer:
+        tactics_list = []
+        techniques_list = []
+        for tactic in tactics or []:
+            if tactic_found := self.get_tactic(tactic=tactic.lower()):
+                tactics_list.append(tactic_found)
+        for technique in techniques or []:
+            if technique_found := self.get_technique(technique_id=technique.lower()):
+                techniques_list.append(technique_found)
+        return MitreInfoContainer(
+            tactics=sorted(tactics_list, key=lambda x: x.name),
+            techniques=sorted(techniques_list, key=lambda x: x.technique_id),
+        )

uncoder-core/app/translator/core/mixins/rule.py

@@ -5,10 +5,12 @@
 import yaml
 
 from app.translator.core.exceptions.core import InvalidJSONStructure, InvalidXMLStructure, InvalidYamlStructure
-from app.translator.core.mitre import MitreConfig
+from app.translator.core.mitre import MitreConfig, MitreInfoContainer
 
 
 class JsonRuleMixin:
+    mitre_config: MitreConfig = MitreConfig()
+
     @staticmethod
     def load_rule(text: str) -> dict:
         try:
@@ -27,18 +29,18 @@ def load_rule(text: str) -> dict:
         except yaml.YAMLError as err:
             raise InvalidYamlStructure(error=str(err)) from err
 
-    def parse_mitre_attack(self, tags: list[str]) -> dict[str, list]:
-        result = {"tactics": [], "techniques": []}
+    def parse_mitre_attack(self, tags: list[str]) -> MitreInfoContainer:
+        parsed_techniques = []
+        parsed_tactics = []
         for tag in set(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)
-            elif tactic := self.mitre_config.get_tactic(tag):
-                result["tactics"].append(tactic)
-        return result
+                parsed_techniques.append(tag)
+            else:
+                parsed_tactics.append(tag)
+        return self.mitre_config.get_mitre_info(tactics=parsed_tactics, techniques=parsed_techniques)
 
 
 class XMLRuleMixin:

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

@@ -1,6 +1,6 @@
 import uuid
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Optional
 
 from app.translator.core.const import QUERY_TOKEN_TYPE
@@ -10,43 +10,80 @@
 from app.translator.core.models.query_tokens.field import Field
 
 
+@dataclass
+class MitreTechniqueContainer:
+    technique_id: str
+    name: str
+    url: str
+    tactic: list[str]
+
+
+@dataclass
+class MitreTacticContainer:
+    external_id: str
+    url: str
+    name: str
+
+
+@dataclass
+class MitreInfoContainer:
+    tactics: list[MitreTacticContainer] = field(default_factory=list)
+    techniques: list[MitreTechniqueContainer] = field(default_factory=list)
+
+
 class MetaInfoContainer:
     def __init__(
         self,
         *,
         id_: Optional[str] = None,
         title: Optional[str] = None,
         description: Optional[str] = None,
-        author: Optional[str] = None,
+        author: Optional[list[str]] = None,
         date: Optional[str] = None,
         output_table_fields: Optional[list[Field]] = None,
         query_fields: Optional[list[Field]] = None,
         license_: Optional[str] = None,
         severity: Optional[str] = None,
         references: Optional[list[str]] = None,
         tags: Optional[list[str]] = None,
-        mitre_attack: Optional[dict[str, list]] = None,
+        raw_mitre_attack: Optional[list[str]] = None,
         status: Optional[str] = None,
         false_positives: Optional[list[str]] = None,
         source_mapping_ids: Optional[list[str]] = None,
         parsed_logsources: Optional[dict] = None,
+        timeframe: Optional[timedelta] = None,
+        mitre_attack: MitreInfoContainer = MitreInfoContainer(),
     ) -> None:
         self.id = id_ or str(uuid.uuid4())
         self.title = title or ""
         self.description = description or ""
-        self.author = author or ""
+        self.author = [v.strip() for v in author] if author else []
         self.date = date or datetime.now().date().strftime("%Y-%m-%d")
         self.output_table_fields = output_table_fields or []
         self.query_fields = query_fields or []
         self.license = license_ or "DRL 1.1"
         self.severity = severity or SeverityType.low
         self.references = references or []
         self.tags = tags or []
-        self.mitre_attack = mitre_attack or {}
+        self.mitre_attack = mitre_attack or None
+        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 = source_mapping_ids or [DEFAULT_MAPPING_NAME]
         self.parsed_logsources = parsed_logsources or {}
+        self.timeframe = timeframe
+
+    @property
+    def author_str(self) -> str:
+        return ", ".join(self.author)
+
+    @property
+    def source_mapping_ids(self) -> list[str]:
+        return sorted(self._source_mapping_ids)
+
+    @source_mapping_ids.setter
+    def source_mapping_ids(self, source_mapping_ids: list[str]) -> None:
+        self._source_mapping_ids = source_mapping_ids
 
 
 @dataclass

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

@@ -77,9 +77,10 @@ def get_field_tokens(
         return field_tokens
 
     def get_source_mappings(
-        self, field_tokens: list[Field], log_sources: dict[str, Union[str, list[str]]]
+        self, field_tokens: list[Field], log_sources: dict[str, list[Union[int, str]]]
     ) -> list[SourceMapping]:
         field_names = [field.source_name for field in field_tokens]
-        source_mappings = self.mappings.get_suitable_source_mappings(field_names=field_names, **log_sources)
+        source_mappings = self.mappings.get_suitable_source_mappings(field_names=field_names, log_sources=log_sources)
         self.tokenizer.set_field_tokens_generic_names_map(field_tokens, source_mappings, self.mappings.default_mapping)
         return source_mappings
+

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

@@ -208,15 +208,17 @@ def wrap_with_not_supported_functions(self, query: str, not_supported_functions:
         return query
 
     def wrap_with_unmapped_fields(self, query: str, fields: Optional[list[str]]) -> str:
-        if fields:
+        if wrap_query_with_meta_info_ctx_var.get() and fields:
             return query + "\n\n" + self.wrap_with_comment(f"{self.unmapped_fields_text}{', '.join(fields)}")
         return query
 
     def wrap_with_comment(self, value: str) -> str:
         return f"{self.comment_symbol} {value}"
 
     @abstractmethod
-    def generate(self, query_container: Union[RawQueryContainer, TokenizedQueryContainer]) -> str:
+    def generate(
+        self, raw_query_container: RawQueryContainer, tokenized_query_container: Optional[TokenizedQueryContainer]
+    ) -> str:
         raise NotImplementedError("Abstract method")
 
 
@@ -318,7 +320,7 @@ def wrap_with_meta_info(self, query: str, meta_info: Optional[MetaInfoContainer]
             meta_info_dict = {
                 "name: ": meta_info.title,
                 "uuid: ": meta_info.id,
-                "author: ": meta_info.author if meta_info.author else "not defined in query/rule",
+                "author: ": meta_info.author_str or "not defined in query/rule",
                 "licence: ": meta_info.license,
             }
             query_meta_info = "\n".join(
@@ -370,7 +372,7 @@ def finalize(self, queries_map: dict[str, str]) -> str:
 
         return result
 
-    def _get_source_mappings(self, source_mapping_ids: list[str]) -> list[SourceMapping]:
+    def _get_source_mappings(self, source_mapping_ids: list[str]) -> Optional[list[SourceMapping]]:
         source_mappings = []
         for source_mapping_id in source_mapping_ids:
             if source_mapping := self.mappings.get_source_mapping(source_mapping_id):
@@ -468,8 +470,9 @@ def generate_from_tokenized_query_container(self, query_container: TokenizedQuer
             raise errors[0]
         return self.finalize(queries_map)
 
-    def generate(self, query_container: Union[RawQueryContainer, TokenizedQueryContainer]) -> str:
-        if isinstance(query_container, RawQueryContainer):
-            return self.generate_from_raw_query_container(query_container)
-
-        return self.generate_from_tokenized_query_container(query_container)
+    def generate(
+        self, raw_query_container: RawQueryContainer, tokenized_query_container: Optional[TokenizedQueryContainer]
+    ) -> str:
+        if tokenized_query_container:
+            return self.generate_from_tokenized_query_container(tokenized_query_container)
+        return self.generate_from_raw_query_container(raw_query_container)

uncoder-core/app/translator/mappings/platforms/qradar/default.yml

@@ -14,8 +14,6 @@ field_mapping:
     - DstPort
     - DestinationPort
     - remoteport
-  dst-hostname: DstHost
-  src-hostname: SrcHost
   src-port:
     - SourcePort
     - localport
@@ -25,13 +23,15 @@ field_mapping:
     - source_ip
     - SourceIP
     - sourceIP
+    - SrcHost
   dst-ip:
     - DestinationIP
     - destinationip
     - destination_ip
     - destinationIP
     - destinationaddress
     - destination
+    - DstHost
   User:
     - userName
     - EventUserName

uncoder-core/app/translator/mappings/utils/load_from_files.py

@@ -6,6 +6,7 @@
 from app.translator.const import APP_PATH
 
 COMMON_FIELD_MAPPING_FILE_NAME = "common.yml"
+DEFAULT_FIELD_MAPPING_FILE_NAME = "default.yml"
 
 
 class LoaderFileMappings:
@@ -23,8 +24,9 @@ def load_mapping(mapping_file_path: str) -> dict:
     def load_platform_mappings(self, platform_dir: str) -> Generator[dict, None, None]:
         platform_path = os.path.join(self.base_mapping_filepath, platform_dir)
         for mapping_file in os.listdir(platform_path):
-            if mapping_file != COMMON_FIELD_MAPPING_FILE_NAME:
+            if mapping_file not in (COMMON_FIELD_MAPPING_FILE_NAME, DEFAULT_FIELD_MAPPING_FILE_NAME):
                 yield self.load_mapping(mapping_file_path=os.path.join(platform_path, mapping_file))
+        yield self.load_mapping(mapping_file_path=os.path.join(platform_path, DEFAULT_FIELD_MAPPING_FILE_NAME))
 
     def load_common_mapping(self, platform_dir: str) -> dict:
         platform_path = os.path.join(self.base_mapping_filepath, platform_dir)

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

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from app.translator.core.mapping import DEFAULT_MAPPING_NAME, BasePlatformMappings, LogSourceSignature, SourceMapping
+from app.translator.core.mapping import BasePlatformMappings, LogSourceSignature
 from app.translator.platforms.athena.const import athena_query_details
 
 
@@ -22,23 +22,5 @@ def prepare_log_source_signature(self, mapping: dict) -> AthenaLogSourceSignatur
         default_log_source = mapping["default_log_source"]
         return AthenaLogSourceSignature(tables=tables, default_source=default_log_source)
 
-    def get_suitable_source_mappings(self, field_names: list[str], table: Optional[str]) -> 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: AthenaLogSourceSignature = source_mapping.log_source_signature
-            if table and log_source_signature.is_suitable(table=table):
-                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:
-            suitable_source_mappings = [self._source_mappings[DEFAULT_MAPPING_NAME]]
-
-        return suitable_source_mappings
-
 
 athena_query_mappings = AthenaMappings(platform_dir="athena", platform_details=athena_query_details)