wip: [stix2 import] Better way to store matchings between indicators and observable objects

Author: chrisr3d

misp_stix_converter/stix2misp/converters/stix2_observable_converter.py

@@ -225,6 +225,10 @@ def x509_hashes_mapping(cls, field: str) -> Union[dict, None]:
 
 
 class STIX2ObservableConverter(STIX2Converter):
+    @property
+    def indicator_references(self) -> dict:
+        return self.main_parser.indicator_references
+
     def _fetch_observable(self, object_ref: str) -> dict:
         return self.main_parser._observable.get(object_ref)
 

misp_stix_converter/stix2misp/converters/stix2_observed_data_converter.py

@@ -58,9 +58,16 @@ class STIX2ObservedDataConverter(STIX2ObservableConverter, metaclass=ABCMeta):
     def __init__(self, main: _MAIN_PARSER_TYPING):
         self._set_main_parser(main)
 
-    def _get_observed_data(
-            self, object_id: str) -> dict | _OBSERVED_DATA_TYPING:
-        return self.main_parser._observed_data[object_id]
+    def _get_observed_data(self, object_id: str) -> _OBSERVED_DATA_TYPING:
+        observed_data = self.main_parser._observed_data[object_id]
+        if isinstance(observed_data, (ObservedData_v20, ObservedData_v21)):
+            return observed_data
+        return observed_data['observed_data']
+
+    def _get_observed_data_indicator_refs(
+            self, observed_data_id: str, object_id: str) -> Union[set, None]:
+        observed_data = self.main_parser._observed_data[observed_data_id]
+        return observed_data.get('indicator_refs', {}).get(object_id)
 
     @property
     def observables(self) -> dict:
@@ -151,31 +158,15 @@ def referenced_ids(self):
 
     def parse(self, observed_data_ref: str):
         observed_data = self._get_observed_data(observed_data_ref)
-        if observed_data.get('indicator_refs') is not None:
-            observed_data_object = observed_data['observed_data']
-            try:
-                if hasattr(observed_data_object, 'object_refs'):
-                    self._parse_observable_object_refs(
-                        observed_data_object, observed_data['indicator_refs']
-                    )
-                else:
-                    self._parse_observable_objects(
-                        observed_data_object, observed_data['indicator_refs']
-                    )
-            except UnknownObservableMappingError as observable_types:
-                self._observable_mapping_error(
-                    observed_data_object.id, observable_types
-                )
-        else:
-            try:
-                if hasattr(observed_data, 'object_refs'):
-                    self._parse_observable_object_refs(observed_data)
-                else:
-                    self._parse_observable_objects(observed_data)
-            except UnknownObservableMappingError as observable_types:
-                self._observable_mapping_error(
-                    observed_data.id, observable_types
-                )
+        try:
+            if hasattr(observed_data, 'object_refs'):
+                self._parse_observable_object_refs(observed_data)
+            else:
+                self._parse_observable_objects(observed_data)
+        except UnknownObservableMappingError as observable_types:
+            self._observable_mapping_error(
+                observed_data.id, observable_types
+            )
 
     def parse_relationships(self):
         for misp_object in self.main_parser.misp_event.objects:

misp_stix_converter/stix2misp/external_stix2_to_misp.py

@@ -25,11 +25,11 @@
 from typing import Iterator, Optional, Union
 
 _EXTENSION_TYPES = (Extension_v20, Extension_v21, STIXBase_v20, STIXBase_v21)
-_INDICATOR_FIELDS = {'_indicator': 1, '_observable': 2, '_observed_data': 4}
 _OBSERVABLE_FIELDS_TO_SKIP = (
     'defanged', 'granular_markings', 'id', 'object_marking_refs',
     'spec_version', 'type'
 )
+_SDO_STORAGE_FIELDS = {'_indicator': 1, '_observable': 2, '_observed_data': 4}
 _SIGHTING_TYPING = Union[Sighting_v20, Sighting_v21]
 
 
@@ -371,23 +371,23 @@ def _fetch_observable_references(
                     if isinstance(value, _EXTENSION_TYPES):
                         yield from self._fetch_observable_references(value)
                         continue
-                    yield f'{key} - {value}'
+                    yield value
                 continue
             if isinstance(values, _EXTENSION_TYPES):
                 yield from self._fetch_observable_references(values)
                 continue
-            yield f'{key} - {values}'
+            yield values
 
     def _set_indicator_references(self):
         score = 0
-        for feature, count in _INDICATOR_FIELDS.items():
+        for feature, count in _SDO_STORAGE_FIELDS.items():
             if getattr(self, feature, []):
                 score += count
         if score in (0, 1, 2, 4, 6):
             return
         pattern_parser = self.indicator_parser._compile_stix_pattern
         self._indicator_references = {
-            indicator_id: tuple(f'{val[0][-1]} - {val[-1]}' for val in pattern)
+            indicator_id: tuple(val[-1] for val in pattern)
             for indicator_id, indicator in self._indicator.items()
             for pattern in pattern_parser(indicator).comparisons.values()
         }
@@ -400,31 +400,44 @@ def _set_indicator_references(self):
                     continue
                 for indicator_reference in indicator_references:
                     indicator = self._indicator[indicator_reference]
-                    self._indicator[indicator_reference] = {
-                        'indicator': indicator,
-                        'observable_ref': observable_id
-                    }
-                    observable['indicator_ref'] = indicator_reference
+                    if not isinstance(indicator, dict):
+                        self._indicator[indicator_reference] = {
+                            'indicator': indicator,
+                            'observable_ref': set()
+                        }
+                        indicator = self._indicator[indicator_reference]
+                    indicator['observable_ref'].add(observable_id)
+                    if 'indicator_ref' not in observable:
+                        observable['indicator_ref'] = {indicator_reference}
+                        continue
+                    observable['indicator_ref'].add(indicator_reference)
         if score >= 5:
             for observed_id, observed_data in self._observed_data.items():
                 if not hasattr(observed_data, 'objects'):
                     continue
-                indicator_refs = {}
+                indicator_refs = defaultdict(set)
                 for observable_id, observable in observed_data.objects.items():
                     indicator_references = set(
                         self._fetch_indicator_reference(observable)
                     )
                     if not indicator_references:
                         continue
                     for indicator_reference in indicator_references:
+                        indicator_refs[observable_id].add(indicator_reference)
                         indicator = self._indicator[indicator_reference]
-                        self._indicator[indicator_reference] = {
-                            'indicator': indicator,
-                            'observable_ref': observed_id
-                        }
-                        indicator_refs[observable_id] = indicator_reference
+                        if not isinstance(indicator, dict):
+                            self._indicator[indicator_reference] = {
+                                'indicator': indicator,
+                                'observable_ref': {observed_id}
+                            }
+                            continue
+                        indicator['observable_ref'].add(observed_id)
                 if indicator_refs:
-                    self._observed_data[observed_id] = {
-                        'indicator_refs': indicator_refs,
-                        'observed_data': observed_data
-                    }
+                    if not isinstance(observed_data, dict):
+                        self._observed_data[observed_id] = {
+                            'indicator_refs': indicator_refs,
+                            'observed_data': observed_data
+                        }
+                        continue
+                    for observable_id, refs in indicator_refs.items():
+                        observed_data['indicator_refs'][observable_id].update(refs)