Changed queue to use JSON serialization (#5032)

Author: joachimmetz

config/dpkg/python3-plaso.install

@@ -3,4 +3,4 @@ usr/lib/python3*/dist-packages/plaso/*/*.py
 usr/lib/python3*/dist-packages/plaso/*/*.yaml
 usr/lib/python3*/dist-packages/plaso/*/*/*.py
 usr/lib/python3*/dist-packages/plaso/*/*/*.yaml
-usr/lib/python3*/dist-packages/plaso*.egg-info
+usr/lib/python3*/dist-packages/plaso*.dist-info

plaso/containers/events.py

@@ -209,6 +209,12 @@ class EventData(interface.AttributeContainer):
 
   CONTAINER_TYPE = 'event_data'
 
+  SCHEMA = {
+      '_event_data_stream_identifier': 'AttributeContainerIdentifier',
+      '_event_values_hash': 'str',
+      '_parser_chain': 'str',
+      'data_type': 'str'}
+
   _SERIALIZABLE_PROTECTED_ATTRIBUTES = [
       '_event_data_stream_identifier',
       '_event_values_hash',
@@ -499,6 +505,25 @@ def SetEventIdentifier(self, event_identifier):
     self._event_identifier = event_identifier
 
 
+class EventTripple(interface.AttributeContainer):
+  """Event tripple.
+
+  Attributes:
+    event (EventObject): event.
+    event_data (EventData): event data.
+    event_data_stream (EventDataStream): event data stream.
+  """
+
+  CONTAINER_TYPE = 'event_tripple'
+
+  def __init__(self):
+    """Initializes an event tripple."""
+    super(EventTripple, self).__init__()
+    self.event = None
+    self.event_data = None
+    self.event_data_stream = None
+
+
 # TODO: the YearLessLogHelper attribute container is kept for backwards
 # compatibility remove once storage format 20230327 is obsolete.
 class YearLessLogHelper(interface.AttributeContainer):
@@ -557,4 +582,4 @@ def SetEventDataStreamIdentifier(self, event_data_stream_identifier):
 
 manager.AttributeContainersManager.RegisterAttributeContainers([
     DateLessLogHelper, EventData, EventDataStream, EventObject, EventTag,
-    YearLessLogHelper])
+    EventTripple, YearLessLogHelper])

plaso/multi_process/analysis_engine.py

@@ -142,7 +142,12 @@ def _AnalyzeEvents(self, storage_writer, analysis_plugins, event_filter=None):
 
       for event_queue in self._event_queues.values():
         # TODO: Check for premature exit of analysis plugins.
-        event_queue.PushItem((event, event_data, event_data_stream))
+        event_tripple = events.EventTripple()
+        event_tripple.event = event
+        event_tripple.event_data = event_data
+        event_tripple.event_data_stream = event_data_stream
+
+        event_queue.PushItem(event_tripple)
 
       self._number_of_consumed_events += 1
 

plaso/multi_process/analysis_process.py

@@ -165,7 +165,7 @@ def _Main(self):
           logger.debug('ConsumeItems exiting, dequeued QueueAbort object.')
           break
 
-        self._ProcessEvent(self._analysis_mediator, *queued_object)
+        self._ProcessEventTripple(self._analysis_mediator, queued_object)
 
         self._number_of_consumed_events += 1
 
@@ -230,19 +230,18 @@ def _Main(self):
     except errors.QueueAlreadyClosed:
       logger.error('Queue for {0:s} was already closed.'.format(self.name))
 
-  def _ProcessEvent(self, mediator, event, event_data, event_data_stream):
-    """Processes an event.
+  def _ProcessEventTripple(self, mediator, event_tripple):
+    """Processes an event tripple.
 
     Args:
       mediator (AnalysisMediator): mediates interactions between
           analysis plugins and other components, such as storage and dfvfs.
-      event (EventObject): event.
-      event_data (EventData): event data.
-      event_data_stream (EventDataStream): event data stream.
+      event_tripple (EventTripple): event tripple.
     """
     try:
       self._analysis_plugin.ExamineEvent(
-          mediator, event, event_data, event_data_stream)
+          mediator, event_tripple.event, event_tripple.event_data,
+          event_tripple.event_data_stream)
 
     except Exception as exception:  # pylint: disable=broad-except
       # TODO: write analysis error and change logger to debug only.

plaso/serializer/json_serializer.py

@@ -65,15 +65,21 @@ def _ConvertAttributeContainerToJSON(cls, attribute_container):
       dict[str, object]: JSON serialized objects.
     """
     if attribute_container.CONTAINER_TYPE not in (
-        'event_data', 'system_configuration'):
+        'event_data', 'event_tripple', 'system_configuration'):
       return cls.ConvertAttributeContainerToJSON(attribute_container)
 
     json_dict = {
         '__type__': 'AttributeContainer',
         '__container_type__': attribute_container.CONTAINER_TYPE}
 
     for attribute_name, attribute_value in attribute_container.GetAttributes():
-      json_dict[attribute_name] = cls._ConvertValueToJSON(attribute_value)
+      if isinstance(
+          attribute_value, containers_interface.AttributeContainerIdentifier):
+        attribute_value = attribute_value.CopyToString()
+      else:
+        attribute_value = cls._ConvertValueToJSON(attribute_value)
+
+      json_dict[attribute_name] = attribute_value
 
     return json_dict
 
@@ -148,7 +154,8 @@ def _ConvertJSONToAttributeContainer(cls, json_dict):
     # Use __container_type__ to indicate the attribute container type.
     container_type = json_dict.get('__container_type__', None)
 
-    if container_type not in ('event_data', 'system_configuration'):
+    if container_type not in (
+        'event_data', 'event_tripple', 'system_configuration'):
       return cls.ConvertJSONToAttributeContainer(json_dict)
 
     attribute_container = cls._CONTAINERS_MANAGER.CreateAttributeContainer(

plaso/storage/sqlite/sqlite_file.py

@@ -75,7 +75,7 @@ def _CreateAttributeContainerFromRow(
       AttributeContainer: attribute container.
     """
     schema = self._GetAttributeContainerSchema(container_type)
-    if schema:
+    if schema and container_type != self._CONTAINER_TYPE_EVENT_DATA:
       return super(SQLiteStorageFile, self)._CreateAttributeContainerFromRow(
           container_type, column_names, row, first_column_index)
 
@@ -106,7 +106,7 @@ def _CreateAttributeContainerTable(self, container_type):
           an unsupported attribute container is provided.
     """
     schema = self._GetAttributeContainerSchema(container_type)
-    if schema:
+    if schema and container_type != self._CONTAINER_TYPE_EVENT_DATA:
       super(SQLiteStorageFile, self)._CreateAttributeContainerTable(
           container_type)
     else:
@@ -276,7 +276,7 @@ def _WriteNewAttributeContainer(self, container):
       OSError: when there is an error querying the storage file.
     """
     schema = self._GetAttributeContainerSchema(container.CONTAINER_TYPE)
-    if schema:
+    if schema and container.CONTAINER_TYPE != self._CONTAINER_TYPE_EVENT_DATA:
       super(SQLiteStorageFile, self)._WriteNewAttributeContainer(container)
     else:
       next_sequence_number = self._GetAttributeContainerNextSequenceNumber(
@@ -328,7 +328,7 @@ def GetAttributeContainerByIndex(self, container_type, index):
           the storage file.
     """
     schema = self._GetAttributeContainerSchema(container_type)
-    if schema:
+    if schema and container_type != self._CONTAINER_TYPE_EVENT_DATA:
       container = super(SQLiteStorageFile, self).GetAttributeContainerByIndex(
           container_type, index)
 
@@ -401,7 +401,7 @@ def GetAttributeContainers(self, container_type, filter_expression=None):
       OSError: when there is an error querying the storage file.
     """
     schema = self._GetAttributeContainerSchema(container_type)
-    if schema:
+    if schema and container_type != self._CONTAINER_TYPE_EVENT_DATA:
       for container in super(SQLiteStorageFile, self).GetAttributeContainers(
           container_type, filter_expression=filter_expression):
         # TODO: the YearLessLogHelper attribute container is kept for backwards

tests/serializer/json_serializer.py

@@ -29,6 +29,8 @@
 class JSONSerializerTestCase(shared_test_lib.BaseTestCase):
   """Tests for a JSON serializer object."""
 
+  # pylint: disable=protected-access
+
   def _TestReadSerialized(self, serializer_object, json_dict):
     """Tests the ReadSerialized function.
 
@@ -78,6 +80,76 @@ class JSONAttributeContainerSerializerTest(JSONSerializerTestCase):
 
   # pylint: disable=protected-access
 
+  def CreateTestEventData(self, event_data_stream):
+    """Creates a test event data.
+
+    Args:
+      event_data_stream (EventDataStream): event data stream.
+
+    Returns:
+      EventData: event data.
+    """
+    event_data = events.EventData()
+    event_data._ignored = 'Not serialized'
+    event_data._parser_chain = 'test_parser'
+    event_data.data_type = 'test:event2'
+
+    event_data.a_tuple = ('some item', [234, 52, 15])
+    event_data.empty_string = ''
+    event_data.float = -122.082203542683
+    event_data.integer = 34
+    event_data.my_list = ['asf', 4234, 2, 54, 'asf']
+    event_data.null_value = None
+    event_data.string = 'Normal string'
+    event_data.unicode_string = 'And I am a unicorn.'
+    event_data.zero_integer = 0
+
+    event_data_stream_identifier = event_data_stream.GetIdentifier()
+    event_data.SetEventDataStreamIdentifier(event_data_stream_identifier)
+
+    return event_data
+
+  def CreateTestEventDataStream(self):
+    """Creates a test event data stream.
+
+    Returns:
+      EventDataSteam: event data stream.
+    """
+    test_file_path = self._GetTestFilePath(['ímynd.dd'])
+
+    volume_path_spec = path_spec_factory.Factory.NewPathSpec(
+        dfvfs_definitions.TYPE_INDICATOR_OS, location=test_file_path)
+    test_path_spec = path_spec_factory.Factory.NewPathSpec(
+        dfvfs_definitions.TYPE_INDICATOR_TSK, location='/',
+        parent=volume_path_spec)
+
+    event_data_stream = events.EventDataStream()
+    event_data_stream.md5_hash = 'e3df0d2abd2c27fbdadfb41a47442520'
+    event_data_stream.path_spec = test_path_spec
+
+    return event_data_stream
+
+  def CreateTestEventObject(self, event_data):
+    """Creates a test event object.
+
+    Args:
+      event_data (EventData): event data.
+
+    Returns:
+      EventObject: an event object.
+    """
+    test_date_time = dfdatetime_posix_time.PosixTime(timestamp=1621839644)
+
+    event = events.EventObject()
+    event.date_time = test_date_time
+    event.timestamp = 1621839644
+    event.timestamp_desc = definitions.TIME_DESCRIPTION_MODIFICATION
+
+    event_data_identifier = event_data.GetIdentifier()
+    event.SetEventDataIdentifier(event_data_identifier)
+
+    return event
+
   def testReadAndWriteSerializedAnalysisReport(self):
     """Test ReadSerialized and WriteSerialized of AnalysisReport."""
     expected_report_text = (
@@ -110,21 +182,8 @@ def testReadAndWriteSerializedAnalysisReport(self):
 
   def testReadAndWriteSerializedEventData(self):
     """Test ReadSerialized and WriteSerialized of EventData."""
-    expected_event_data = events.EventData()
-    expected_event_data._event_data_stream_identifier = 'event_data_stream.1'
-    expected_event_data._ignored = 'Not serialized'
-    expected_event_data._parser_chain = 'test_parser'
-    expected_event_data.data_type = 'test:event2'
-
-    expected_event_data.a_tuple = ('some item', [234, 52, 15])
-    expected_event_data.empty_string = ''
-    expected_event_data.float = -122.082203542683
-    expected_event_data.integer = 34
-    expected_event_data.my_list = ['asf', 4234, 2, 54, 'asf']
-    expected_event_data.null_value = None
-    expected_event_data.string = 'Normal string'
-    expected_event_data.unicode_string = 'And I am a unicorn.'
-    expected_event_data.zero_integer = 0
+    expected_event_data_stream = self.CreateTestEventDataStream()
+    expected_event_data = self.CreateTestEventData(expected_event_data_stream)
 
     json_string = (
         json_serializer.JSONAttributeContainerSerializer.WriteSerialized(
@@ -140,7 +199,8 @@ def testReadAndWriteSerializedEventData(self):
     self.assertIsInstance(event_data, events.EventData)
 
     expected_event_data_dict = {
-        '_event_data_stream_identifier': 'event_data_stream.1',
+        '_event_data_stream_identifier': (
+            expected_event_data.GetEventDataStreamIdentifier().CopyToString()),
         '_parser_chain': 'test_parser',
         'a_tuple': ('some item', [234, 52, 15]),
         'data_type': 'test:event2',
@@ -157,17 +217,7 @@ def testReadAndWriteSerializedEventData(self):
 
   def testReadAndWriteSerializedEventDataStream(self):
     """Test ReadSerialized and WriteSerialized of EventDataStream."""
-    test_file_path = self._GetTestFilePath(['ímynd.dd'])
-
-    volume_path_spec = path_spec_factory.Factory.NewPathSpec(
-        dfvfs_definitions.TYPE_INDICATOR_OS, location=test_file_path)
-    test_path_spec = path_spec_factory.Factory.NewPathSpec(
-        dfvfs_definitions.TYPE_INDICATOR_TSK, location='/',
-        parent=volume_path_spec)
-
-    expected_event_data_stream = events.EventDataStream()
-    expected_event_data_stream.md5_hash = 'e3df0d2abd2c27fbdadfb41a47442520'
-    expected_event_data_stream.path_spec = test_path_spec
+    expected_event_data_stream = self.CreateTestEventDataStream()
 
     json_string = (
         json_serializer.JSONAttributeContainerSerializer.WriteSerialized(
@@ -184,21 +234,17 @@ def testReadAndWriteSerializedEventDataStream(self):
 
     expected_event_data_stream_dict = {
         'md5_hash': 'e3df0d2abd2c27fbdadfb41a47442520',
-        'path_spec': test_path_spec}
+        'path_spec': expected_event_data_stream.path_spec}
 
     event_data_stream_dict = event_data_stream.CopyToDict()
 
     self.assertEqual(event_data_stream_dict, expected_event_data_stream_dict)
 
   def testReadAndWriteSerializedEventObject(self):
     """Test ReadSerialized and WriteSerialized of EventObject."""
-    test_date_time = dfdatetime_posix_time.PosixTime(timestamp=1621839644)
-
-    expected_event = events.EventObject()
-    expected_event._event_data_identifier = 'event_data.1'
-    expected_event.date_time = test_date_time
-    expected_event.timestamp = 1621839644
-    expected_event.timestamp_desc = definitions.TIME_DESCRIPTION_MODIFICATION
+    expected_event_data_stream = self.CreateTestEventDataStream()
+    expected_event_data = self.CreateTestEventData(expected_event_data_stream)
+    expected_event = self.CreateTestEventObject(expected_event_data)
 
     json_string = (
         json_serializer.JSONAttributeContainerSerializer.WriteSerialized(
@@ -213,8 +259,9 @@ def testReadAndWriteSerializedEventObject(self):
     self.assertIsInstance(event, events.EventObject)
 
     expected_event_dict = {
-        '_event_data_identifier': 'event_data.1',
-        'date_time': test_date_time.CopyToDateTimeStringISO8601(),
+        '_event_data_identifier': (
+            expected_event.GetEventDataIdentifier().CopyToString()),
+        'date_time': expected_event.date_time.CopyToDateTimeStringISO8601(),
         'timestamp': 1621839644,
         'timestamp_desc': definitions.TIME_DESCRIPTION_MODIFICATION}
 
@@ -296,6 +343,29 @@ def testReadAndWriteSerializedEventTag(self):
         sorted(event_tag_dict.items()),
         sorted(expected_event_tag_dict.items()))
 
+  def testReadAndWriteSerializedEventTripple(self):
+    """Test ReadSerialized and WriteSerialized of EventTripple."""
+    expected_event_data_stream = self.CreateTestEventDataStream()
+    expected_event_data = self.CreateTestEventData(expected_event_data_stream)
+    expected_event_tripple = events.EventTripple()
+    expected_event_tripple.event = self.CreateTestEventObject(
+        expected_event_data)
+    expected_event_tripple.event_data = expected_event_data
+    expected_event_tripple.event_data_stream = expected_event_data_stream
+
+    json_string = (
+        json_serializer.JSONAttributeContainerSerializer.WriteSerialized(
+            expected_event_tripple))
+
+    self.assertIsNotNone(json_string)
+
+    event_tripple = (
+        json_serializer.JSONAttributeContainerSerializer.ReadSerialized(
+            json_string))
+
+    self.assertIsNotNone(event_tripple)
+    self.assertIsInstance(event_tripple, events.EventTripple)
+
   def testReadAndWriteSerializedSession(self):
     """Test ReadSerialized and WriteSerialized of Session."""
     expected_session = sessions.Session()