wip: [tests, documentation] Changes on the mapping and summary of the MISP to STIX export documentation

- Supporting inputs taken from test samples which
  are now validated MISP format
- Generating simplified mapping documentation with
  a single STIX result for a given MISP sample
  covering how data is converted depending or not
  on a `to_ids` flag, to better document the
  actual behavior implemented recently to convert
  to both Observable objects and Indicators (and
  the conversion to other types of objects remains
  unchanged)

Author: chrisr3d

tests/update_documentation.py

@@ -2,10 +2,11 @@
 # -*- coding: utf-8 -*-
 
 import json
+from base64 import b64encode
 from pathlib import Path
 
 _ROOT_PATH = Path(__file__).parents[1].resolve()
-_OBJECT_FEATURES = ('Indicator', 'Observed Data')
+_OBJECT_FEATURES = ('indicator', 'observed-data')
 _PATTERNING_TYPES = (
     'sigma',
     'snort',
@@ -26,11 +27,11 @@ def __init__(self, filename, feature):
         documentation_path = _ROOT_PATH / 'documentation' / 'mapping'
         self.__mapping_path = documentation_path / f'{filename}.json'
         with open(self.__mapping_path, 'rt', encoding='utf-8') as f:
-            self._documentation = json.loads(f.read())
+            self._documentation = json.load(f)
         self.__summary_path = documentation_path / f'{filename}_summary.json'
         try:
             with open(self.__summary_path, 'rt', encoding='utf-8') as f:
-                self._summary = json.loads(f.read())
+                self._summary = json.load(f)
         except (FileNotFoundError, json.decoder.JSONDecodeError):
             self._summary = {}
         self._summary_changed = False
@@ -64,10 +65,8 @@ def check_export_mapping(self, feature):
                 else:
                     if mapping['MISP'] != self._documentation[name]['MISP']:
                         self._documentation[name]['MISP'] = mapping['MISP']
-                    for stix_type, stix_object in mapping['STIX'].items():
-                        stixobject = self._documentation[name]['STIX'].get(stix_type, {})
-                        if stix_object != stixobject:
-                            self._documentation[name]['STIX'][stix_type] = stix_object
+                    if mapping['STIX'] != self._documentation[name]['STIX']:
+                        self._documentation[name]['STIX'] = mapping['STIX']
                 self._check_stix_export_summary(name, mapping['STIX'], feature)
             self._write_documentation()
         else:
@@ -115,13 +114,40 @@ def _declare_mapping(self, mapping):
     def _declare_summary(self, summary):
         self.__summary_mapping = summary
 
+    def _define_export_summary(self, *object_types):
+        return ' / '.join(
+            f'**{object_type.capitalize()}**' for object_type in object_types
+        )
+
+    def _define_import_summary(self, stix_mapping):
+        types = []
+        for stix_type, mapping in stix_mapping.items():
+            stix_object = mapping['STIX']
+            if stix_type == 'Indicator':
+                types.append(self._pattern_types(stix_object['pattern']))
+            elif stix_type == 'Observed Data':
+                observables = stix_object[1:] if isinstance(stix_object, list) else tuple(stix_object['objects'].values())
+                types.append(f"{self._observable_types(observables)} (observable)")
+            else:
+                types.append(f'**{stix_type}**')
+        return ' / '.join(types)
+
     def _define_stix20_export_summary(self, stix_mapping):
-        if all(feature in stix_mapping for feature in _OBJECT_FEATURES):
-            return self._observable_types(stix_mapping['Observed Data']['objects'].values())
-        if len(stix_mapping.keys()) == 1 and 'Indicator' in stix_mapping:
-            indicator_type = self._pattern_types(stix_mapping['Indicator']['pattern'])
-            return f"{indicator_type} / Custom Object"
-        return self._define_export_summary(stix_mapping)
+        if isinstance(stix_mapping, dict):
+            return self._define_export_summary(stix_mapping['type'])
+        object_types = {
+            stix_object['type'] for stix_object in stix_mapping
+            if stix_object['type'] != 'relationship'
+        }
+        if all(object_type in object_types for object_type in _OBJECT_FEATURES):
+            return self._observable_types(
+                *(
+                    observable['type'] for stix_object in stix_mapping
+                    if stix_object['type'] == 'observed-data'
+                    for observable in stix_object['objects']
+                )
+            )
+        return self._define_export_summary(*object_types)
 
     def _define_stix20_import_summary(self, stix_mapping):
         if all(feature in stix_mapping for feature in _OBJECT_FEATURES):
@@ -132,14 +158,20 @@ def _define_stix20_import_summary(self, stix_mapping):
         return self._define_import_summary(stix_mapping)
 
     def _define_stix21_export_summary(self, stix_mapping):
-        if all(feature in stix_mapping for feature in _OBJECT_FEATURES):
-            return self._observable_types(stix_mapping['Observed Data'][1:])
-        if len(stix_mapping.keys()) == 1 and 'Indicator' in stix_mapping:
-            indicator = stix_mapping['Indicator']
-            if indicator['pattern_type'] in _PATTERNING_TYPES:
-                return '**Indicator**'
-            return f"{self._pattern_types(indicator['pattern'])} / Custom Object"
-        return self._define_export_summary(stix_mapping)
+        if isinstance(stix_mapping, dict):
+            return self._define_export_summary(stix_mapping['type'])
+        object_types = {
+            stix_object['type'] for stix_object in stix_mapping
+            if stix_object['type'] != 'relationship'
+        }
+        if all(object_type in object_types for object_type in _OBJECT_FEATURES):
+            return self._observable_types(
+                *(
+                    object_type for object_type in object_types
+                    if object_type not in _OBJECT_FEATURES
+                )
+            )
+        return self._define_export_summary(*object_types)
 
     def _define_stix21_import_summary(self, stix_mapping):
         if all(feature in stix_mapping for feature in _OBJECT_FEATURES):
@@ -151,40 +183,23 @@ def _define_stix21_import_summary(self, stix_mapping):
             return f"{self._pattern_types(indicator['pattern'])} / Custom Object"
         return self._define_import_summary(stix_mapping)
 
-    def _define_export_summary(self, stix_mapping):
-        types = []
-        for stix_type, stix_object in stix_mapping.items():
-            if stix_type == 'Indicator':
-                types.append(self._pattern_types(stix_object['pattern']))
-            elif stix_type == 'Observed Data':
-                observables = stix_object[1:] if isinstance(stix_object, list) else tuple(stix_object['objects'].values())
-                types.append(f"{self._observable_types(observables)} (observable)")
-            else:
-                types.append(f'**{stix_type}**')
-        return ' / '.join(types)
-
-    def _define_import_summary(self, stix_mapping):
-        types = []
-        for stix_type, mapping in stix_mapping.items():
-            stix_object = mapping['STIX']
-            if stix_type == 'Indicator':
-                types.append(self._pattern_types(stix_object['pattern']))
-            elif stix_type == 'Observed Data':
-                observables = stix_object[1:] if isinstance(stix_object, list) else tuple(stix_object['objects'].values())
-                types.append(f"{self._observable_types(observables)} (observable)")
-            else:
-                types.append(f'**{stix_type}**')
-        return ' / '.join(types)
-
     @staticmethod
     def _observable_type(observable_type):
         if observable_type in _SUMMARY_MAPPING:
             return _SUMMARY_MAPPING[observable_type]
         return observable_type.replace('-', ' ').title()
 
-    def _observable_types(self, observables):
-        types = {self._observable_type(observable['type']) for observable in observables}
-        return f"{' & '.join(sorted(types))} {'Objects' if len(observables) > 1 else 'Object'}"
+    def _observable_types(self, *observable_types):
+        types = ' & '.join(
+            sorted(
+                set(
+                    ' '.join(observable_type.split('-')).title()
+                    for observable_type in observable_types
+                )
+            )
+        )
+        objects = 'Objects' if len(observable_types) > 1 else 'Object'
+        return f"{types} {objects} and IoCs described in Indicator (pattern)"
 
     def _order_mapping(self, name):
         return {key: attribute for key, attribute in sorted(getattr(self, name).items())}
@@ -195,49 +210,49 @@ def _pattern_types(self, pattern):
             types.add(self._observable_type(part.split(':')[0]))
         return f"{' & '.join(types)} {'Objects' if len(types) > 1 else 'Object'} (pattern)"
 
-    def _replace_data(self, attribute, name, stix_mapping):
-        data = attribute['data']
+    def _replace_data(self, name, misp_mapping, stix_mapping):
+        data = misp_mapping['data']
         short_data = f"{data[:23]}[...]{data[-23:]}"
-        attribute['data'] = short_data
+        misp_mapping['data'] = short_data
         getattr(self, self.data_replacement[name].format(self.feature))(stix_mapping, data, short_data)
 
     @staticmethod
     def _replace_export_file_data(mapping, data, short_data):
-        if mapping.get('Indicator', {}).get('pattern') is not None:
-            mapping['Indicator']['pattern'] = mapping['Indicator']['pattern'].replace(data, short_data)
-        if isinstance(mapping['Observed Data'], list):
-            for observable in mapping['Observed Data'][1:]:
-                if observable['type'] == 'artifact':
-                    observable['payload_bin'] = observable['payload_bin'].replace(data, short_data)
-                    break
-        else:
-            for index, observable in mapping['Observed Data']['objects'].items():
-                if observable['type'] == 'artifact':
-                    mapping['Observed Data']['objects'][index]['payload_bin'] = observable['payload_bin'].replace(data, short_data)
-                    break
+        for stix_object in mapping:
+            if stix_object['type'] == 'indicator':
+                stix_object['pattern'] = stix_object['pattern'].replace(data, short_data)
+                continue
+            if stix_object['type'] == 'observed-data' and 'objects' in stix_object:
+                for index, observable in stix_object['objects'].items():
+                    if observable['type'] == 'artifact':
+                        stix_object['objects'][index]['payload_bin'] = short_data
+                        continue
+                continue
+            if stix_object['type'] == 'artifact':
+                stix_object['payload_bin'] = short_data
 
     @staticmethod
     def _replace_import_file_data(mapping, data, short_data):
         if isinstance(mapping, list):
             for observable in mapping[1:]:
                 if observable['type'] == 'artifact':
-                    observable['payload_bin'] = observable['payload_bin'].replace(data, short_data)
+                    observable['payload_bin'] = short_data
                     break
         else:
             if mapping['type'] == 'indicator':
                 mapping['pattern'] = mapping['pattern'].replace(data, short_data)
             else:
                 for index, observable in mapping['objects'].items():
                     if observable['type'] == 'artifact':
-                        mapping['objects'][index]['payload_bin'] = observable['payload_bin'].replace(data, short_data)
+                        mapping['objects'][index]['payload_bin'] = short_data
 
     def _write_documentation(self):
         with open(self.mapping_path, 'wt', encoding='utf-8') as f:
-            f.write(json.dumps(self._order_mapping('_documentation'), indent=4))
+            f.write(json.dumps(dict(self._order_mapping('_documentation')), indent=4))
 
     def _write_summary(self):
         with open(self.summary_path, 'wt', encoding='utf-8') as f:
-            f.write(json.dumps(self._order_mapping('_summary'), indent=4))
+            f.write(json.dumps(dict(self._order_mapping('_summary')), indent=4))
 
 
 class AttributesDocumentationUpdater(DocumentationUpdater):
@@ -255,7 +270,7 @@ def data_replacement(cls):
 
     def _check_data(self, attribute_type, mapping):
         if 'data' in mapping['MISP'] and len(mapping['MISP']['data']) > 51:
-            self._replace_data(mapping['MISP'], attribute_type, mapping['STIX'])
+            self._replace_data(attribute_type, mapping['MISP'], mapping['STIX'])
 
     def _load_attributes_mapping(self, attributes_mapping):
         self._declare_summary(attributes_mapping.pop('summary', {}))
@@ -307,13 +322,12 @@ def _check_data(self, name, mapping):
         if isinstance(mapping['MISP'], list):
             for misp_object in mapping['MISP']:
                 for attribute in misp_object['Attribute']:
-                    if 'data' in attribute and attribute['data'] is not None and len(attribute['data']) > 51:
-                        self._replace_data(attribute, name, mapping['STIX'])
+                    if 'data' in attribute and len(attribute['data']) > 51:
+                        self._replace_data(name, attribute, mapping['STIX'])
         else:
             for attribute in mapping['MISP']['Attribute']:
                 if 'data' in attribute and len(attribute['data']) > 51:
-                    self._replace_data(attribute, name, mapping['STIX'])
-
+                    self._replace_data(name, attribute, mapping['STIX'])
 
     def _load_objects_mapping(self, objects_mapping):
         self._declare_summary(objects_mapping.pop('summary', {}))
@@ -327,40 +341,47 @@ def _load_objects_mapping(self, objects_mapping):
         self._declare_mapping(objects_mapping)
 
     def _replace_export_account_data(self, mapping, data, short_data):
-        mapping['Indicator']['pattern'] = mapping['Indicator']['pattern'].replace(data, short_data)
-        if isinstance(mapping['Observed Data'], list):
-            for observable in mapping['Observed Data'][1:]:
-                self._replace_custom_field(observable, data, short_data)
-        else:
-            for observable in mapping['Observed Data']['objects'].values():
-                self._replace_custom_field(observable, data, short_data)
+        for stix_object in mapping:
+            if stix_object['type'] == 'indicator':
+                stix_object['pattern'] = stix_object['pattern'].replace(data, short_data)
+                continue
+            if stix_object['type'] == 'observed-data':
+                if 'objects' in stix_object:
+                    for observable in stix_object['objects'].values():
+                        self._replace_custom_field(observable, short_data)
+                continue
+            if stix_object['type'] != 'relationship':
+                self._replace_custom_field(stix_object, short_data)
 
     def _replace_import_account_data(self, mapping, data, short_data):
         if isinstance(mapping, list):
             for observable in mapping[1:]:
-                self._replace_custom_field(observable, data, short_data)
+                self._replace_custom_field(observable, short_data)
         else:
             if mapping['type'] == 'indicator':
                 mapping['pattern'] = mapping['pattern'].replace(data, short_data)
             else:
                 for observable in mapping['objects'].values():
-                    self._replace_custom_field(observable, data, short_data)
+                    self._replace_custom_field(observable, short_data)
 
     @staticmethod
-    def _replace_custom_field(mapping, data, short_data):
+    def _replace_custom_field(mapping, short_data):
         for feature, value in mapping.items():
             if feature.startswith('x_misp_') and isinstance(value, dict):
-                value['data'] = value['data'].replace(data, short_data)
+                value['data'] = short_data
                 break
 
-    def _replace_export_identity_data(self, mapping, data, short_data):
-        self._replace_custom_field(mapping['Identity'], data, short_data)
+    def _replace_export_identity_data(self, mapping, _, short_data):
+        self._replace_custom_field(mapping, short_data)
 
-    def _replace_import_identity_data(self, mapping, data, short_data):
-        self._replace_custom_field(mapping, data, short_data)
+    def _replace_import_identity_data(self, mapping, _, short_data):
+        self._replace_custom_field(mapping, short_data)
 
-    def _replace_export_annotation_data(self, mapping, data, short_data):
-        self._replace_custom_field(mapping['Note'], data, short_data)
+    def _replace_export_annotation_data(self, mapping, _, short_data):
+        for stix_object in mapping:
+            if stix_object['type'] != 'note':
+                continue
+            self._replace_custom_field(stix_object, short_data)
 
-    def _replace_import_annotation_data(self, mapping, data, short_data):
-        self._replace_custom_field(mapping, data, short_data)
+    def _replace_import_annotation_data(self, mapping, _, short_data):
+        self._replace_custom_field(mapping, short_data)