chg: [stix export] Made the parse_json_content method for actual content and added parse_file_content to pass filename(s)

Author: chrisr3d

misp_stix_converter/misp2stix/exportparser.py

@@ -1,15 +1,18 @@
 # -*- coding: utf-8 -*-
 #!/usr/bin/env python3
 
+import json
 import traceback
 from .stix1_mapping import MISPtoSTIX1Mapping
 from .stix20_mapping import MISPtoSTIX20Mapping
 from .stix21_mapping import MISPtoSTIX21Mapping
 from abc import ABCMeta
 from collections import defaultdict
 from datetime import datetime, timezone
+from io import BufferedIOBase, TextIOBase
+from pathlib import Path
 from pymisp import MISPAttribute, MISPObject
-from typing import Optional, Union
+from typing import IO, Optional, Union
 
 
 class MISPtoSTIXParser(metaclass=ABCMeta):
@@ -42,6 +45,18 @@ def warnings(self) -> dict:
             for identifier, warnings in self.__warnings.items()
         }
 
+    def parse_json_file(self, filename: Path | str):
+        with open(filename, 'rt', encoding='utf-8') as f:
+            json_content = json.loads(f.read())
+        self._parse_json_content(json_content)
+
+    def parse_json_content(self, json_content: IO | str | bytes | dict | list):
+        if isinstance(json_content, (BufferedIOBase, TextIOBase)):
+            json_content = json_content.read()
+        if isinstance(json_content, (str, bytes)):
+            json_content = json.loads(json_content)
+        self._parse_json_content(json_content)
+
     ############################################################################
     #                         COMMON PARSING FUNCTIONS                         #
     ############################################################################

misp_stix_converter/misp2stix/misp_to_stix1.py

@@ -1102,11 +1102,9 @@ def __init__(self, orgname: str, version: str):
         self._identifier = 'attributes collection'
         self._ids = set()
 
-    def parse_json_content(self, filename):
-        with open(filename, 'rt', encoding='utf-8') as f:
-            attributes = json.loads(f.read())
-            if attributes.get('response') is not None:
-                attributes = attributes['response']
+    def _parse_json_content(self, attributes: dict | list):
+        if attributes.get('response') is not None:
+            attributes = attributes['response']
         self._stix_package = STIXPackage()
         if 'Attribute' in attributes:
             attributes = attributes['Attribute']
@@ -1152,9 +1150,7 @@ class MISPtoSTIX1EventsParser(MISPtoSTIX1Parser):
     def __init__(self, orgname: str, version: str):
         super().__init__(orgname, version)
 
-    def parse_json_content(self, filename):
-        with open(filename, 'rt', encoding='utf-8') as f:
-            json_content = json.loads(f.read())
+    def _parse_json_content(self, json_content: dict):
         if json_content.get('response'):
             package = _stix_package(self._orgname, self._version, header=False)
             for event in json_content['response']:

misp_stix_converter/misp2stix/misp_to_stix2.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-import io
 import json
 import os
 import re
@@ -10,7 +9,7 @@
 from base64 import b64encode
 from collections import defaultdict
 from datetime import datetime
-from dateutil import parser
+from io import BytesIO
 from pathlib import Path
 from pymisp import (
     MISPAttribute, MISPEvent, MISPEventReport, MISPGalaxy, MISPGalaxyCluster,
@@ -83,31 +82,6 @@ def __init__(self, interoperability: bool):
         }
         self._markings = {}
 
-    def parse_json_content(self, filename: Union[Path, str]):
-        self._results_handling_function = '_append_SDO'
-        with open(filename, 'rt', encoding='utf-8') as f:
-            json_content = json.loads(f.read())
-        if 'response' in json_content:
-            json_content = json_content['response']
-            if isinstance(json_content, list):
-                if not self.__initiated:
-                    self._initiate_events_parsing()
-                for event in json_content:
-                    self._parse_misp_event(event)
-                    self.__index = len(self.__objects)
-            else:
-                self.parse_misp_attributes(json_content)
-        else:
-            if isinstance(json_content, list):
-                for content in json_content:
-                    if 'Attribute' in content:
-                        self.parse_misp_attribute(content)
-            else:
-                if 'Event' in json_content or 'info' in json_content:
-                    self.parse_misp_event(json_content)
-                else:
-                    self.parse_misp_attributes(json_content)
-
     def parse_misp_attribute(self, attribute: Union[MISPAttribute, dict]):
         self._results_handling_function = '_append_SDO_without_refs'
         self._identifier = 'attribute feed'
@@ -143,7 +117,30 @@ def parse_misp_event(self, misp_event: Union[MISPEvent, dict]):
             self._initiate_events_parsing()
         self._parse_misp_event(misp_event)
 
-    def _parse_misp_event(self, misp_event: Union[MISPEvent, dict]):
+    def _parse_json_content(self, json_content: dict | list):
+        self._results_handling_function = '_append_SDO'
+        if 'response' in json_content:
+            json_content = json_content['response']
+            if isinstance(json_content, list):
+                if not self.__initiated:
+                    self._initiate_events_parsing()
+                for event in json_content:
+                    self._parse_misp_event(event)
+                    self.__index = len(self.__objects)
+            else:
+                self.parse_misp_attributes(json_content)
+        else:
+            if isinstance(json_content, list):
+                for content in json_content:
+                    if 'Attribute' in content:
+                        self.parse_misp_attribute(content)
+            else:
+                if 'Event' in json_content or 'info' in json_content:
+                    self.parse_misp_event(json_content)
+                else:
+                    self.parse_misp_attributes(json_content)
+
+    def _parse_misp_event(self, misp_event: MISPEvent | dict):
         if 'Event' in misp_event:
             misp_event = misp_event['Event']
         self._misp_event = misp_event
@@ -173,18 +170,18 @@ def _parse_misp_event(self, misp_event: Union[MISPEvent, dict]):
         self.__objects.insert(self.__index, report)
 
     def _define_stix_object_id(
-            self, feature: str, misp_object: Union[MISPObject, dict]) -> str:
+            self, feature: str, misp_object: MISPObject | dict) -> str:
         return f"{feature}--{misp_object['uuid']}"
 
     def _define_stix_object_id_from_attribute(
-            self, feature: str, attribute: Union[MISPAttribute, dict]) -> str:
+            self, feature: str, attribute: MISPAttribute | dict) -> str:
         attribute_uuid = attribute['uuid']
         stix_id = f'{feature}--{attribute_uuid}'
         self._event_report_matching[attribute_uuid].append(stix_id)
         return stix_id
 
     def _define_stix_object_id_from_object(
-            self, feature: str, misp_object: Union[MISPObject, dict]) -> str:
+            self, feature: str, misp_object: MISPObject | dict) -> str:
         object_uuid = misp_object['uuid']
         stix_id = f'{feature}--{object_uuid}'
         self._event_report_matching[object_uuid].append(stix_id)
@@ -3870,7 +3867,7 @@ def _parse_lnk_args(self, attributes: dict, misp_object) -> dict:
         return file_args
 
     def _parse_malware_sample_additional_fields(
-            self, data: Union[io.BytesIO, str]) -> dict:
+            self, data: BytesIO | str) -> dict:
         if not isinstance(data, str):
             data = b64encode(data.getvalue()).decode()
         return {
@@ -3879,7 +3876,7 @@ def _parse_malware_sample_additional_fields(
         }
 
     def _parse_malware_sample_args(
-            self, value: str, data: Union[io.BytesIO, str]) -> dict:
+            self, value: str, data: BytesIO | str) -> dict:
         args = {'allow_custom': True}
         for separator in self.composite_separators:
             if separator in value:
@@ -3897,7 +3894,7 @@ def _parse_malware_sample_args(
         return args
 
     def _parse_malware_sample_custom_args(
-            self, value: str, data: Union[io.BytesIO, str]) -> dict:
+            self, value: str, data: BytesIO | str) -> dict:
         args = {'allow_custom': True, 'x_misp_malware_sample': value}
         args.update(self._parse_malware_sample_additional_fields(data))
         return args

misp_stix_converter/misp_stix_converter.py

@@ -269,7 +269,7 @@ def misp_attribute_collection_to_stix1(
             filename = input_files[0]
             if isinstance(filename, str):
                 filename = Path(filename).resolve()
-            parser.parse_json_content(filename)
+            parser.parse_json_file(filename)
             name = _check_filename(
                 filename.parent, f'{filename.name}.out', output_dir, output_name
             )
@@ -290,7 +290,7 @@ def misp_attribute_collection_to_stix1(
         if in_memory:
             for filename in input_files:
                 try:
-                    parser.parse_json_content(filename)
+                    parser.parse_json_file(filename)
                     current = parser.stix_package
                     for campaign in current.campaigns:
                         stix_package.add_campaign(campaign)
@@ -319,7 +319,7 @@ def misp_attribute_collection_to_stix1(
         tmp_path = name.parent
         for filename in input_files:
             try:
-                parser.parse_json_content(filename)
+                parser.parse_json_file(filename)
                 package = parser.stix_package
                 for feature in _STIX1_features:
                     values = getattr(package, feature)
@@ -364,7 +364,7 @@ def misp_attribute_collection_to_stix1(
         try:
             if isinstance(filename, str):
                 filename = Path(filename).resolve()
-            parser.parse_json_content(filename)
+            parser.parse_json_file(filename)
             name = _check_output(
                 filename.parent, f'{filename.name}.out', output_dir
             )
@@ -400,7 +400,7 @@ def misp_event_collection_to_stix1(
         try:
             if not isinstance(filename, Path):
                 filename = Path(filename).resolve()
-            parser.parse_json_content(filename)
+            parser.parse_json_file(filename)
             name = _check_filename(
                 filename.parent, f'{filename.name}.out', output_dir, output_name
             )
@@ -421,7 +421,7 @@ def misp_event_collection_to_stix1(
                 try:
                     if not isinstance(filename, Path):
                         filename = Path(filename).resolve()
-                    parser.parse_json_content(filename)
+                    parser.parse_json_file(filename)
                     if parser.stix_package.related_packages is not None:
                         for related_package in parser.stix_package.related_packages:
                             stix_package.add_related_package(related_package)
@@ -440,7 +440,7 @@ def misp_event_collection_to_stix1(
         try:
             if not isinstance(filename, Path):
                 filename = Path(filename).resolve()
-            parser.parse_json_content(filename)
+            parser.parse_json_file(filename)
             content = _get_events(parser.stix_package, return_format)
             with open(name, 'wt', encoding='utf-8') as f:
                 f.write(f'{header}{content}')
@@ -450,7 +450,7 @@ def misp_event_collection_to_stix1(
             try:
                 if not isinstance(filename, Path):
                     filename = Path(filename).resolve()
-                parser.parse_json_content(filename)
+                parser.parse_json_file(filename)
                 content = _get_events(parser.stix_package, return_format)
                 with open(name, 'at', encoding='utf-8') as f:
                     f.write(f'{separator}{content}')
@@ -465,7 +465,7 @@ def misp_event_collection_to_stix1(
         try:
             if not isinstance(filename, Path):
                 filename = Path(filename).resolve()
-            parser.parse_json_content(filename)
+            parser.parse_json_file(filename)
             name = _check_output(
                 filename.parent, f'{filename.name}.out', output_dir
             )
@@ -493,7 +493,7 @@ def misp_collection_to_stix2(
         try:
             if not isinstance(filename, Path):
                 filename = Path(filename).resolve()
-            parser.parse_json_content(filename)
+            parser.parse_json_file(filename)
             name = _check_filename(
                 filename.parent, f'{filename.name}.out', output_dir, output_name
             )
@@ -509,7 +509,7 @@ def misp_collection_to_stix2(
                 try:
                     if not isinstance(filename, Path):
                         filename = Path(filename).resolve()
-                    parser.parse_json_content(filename)
+                    parser.parse_json_file(filename)
                 except Exception as exception:
                     traceback['fails'].append(f'{filename} - {exception.__str__()}')
             if any(filename not in traceback.get('fails', []) for filename in input_files):
@@ -537,7 +537,7 @@ def misp_collection_to_stix2(
             filename = input_files[0]
             if not isinstance(filename, Path):
                 filename = Path(filename).resolve()
-            parser.parse_json_content(filename)
+            parser.parse_json_file(filename)
             stix_objects = json.dumps(
                 [parser.fetch_stix_objects], cls=STIXJSONEncoder, indent=4
             )
@@ -550,7 +550,7 @@ def misp_collection_to_stix2(
             try:
                 if not isinstance(filename, Path):
                     filename = Path(filename).resolve()
-                parser.parse_json_content(filename)
+                parser.parse_json_file(filename)
                 stix_objects = json.dumps(
                     [parser.fetch_stix_objects], cls=STIXJSONEncoder, indent=4
                 )
@@ -572,7 +572,7 @@ def misp_collection_to_stix2(
         try:
             if not isinstance(filename, Path):
                 filename = Path(filename).resolve()
-            parser.parse_json_content(filename)
+            parser.parse_json_file(filename)
             name = _check_output(
                 filename.parent, f'{filename.name}.out', output_dir
             )
@@ -602,7 +602,7 @@ def misp_to_stix1(
     try:
         if not isinstance(filename, Path):
             filename = Path(filename).resolve()
-        parser.parse_json_content(filename)
+        parser.parse_json_file(filename)
         name = _check_filename(
             filename.parent, f'{filename.name}.out', output_dir, output_name
         )