Source code for plaso.parsers.olecf

"""Parser for OLE Compound Files (OLECF)."""

import pyolecf

from plaso.lib import specification
from plaso.parsers import interface
from plaso.parsers import logger
from plaso.parsers import manager




class OLECFParser(interface.FileObjectParser):
  """Parses OLE Compound Files (OLECF)."""

  # pylint: disable=no-member

  NAME = 'olecf'
  DATA_FORMAT = 'OLE Compound File (OLECF) format'

  _INITIAL_FILE_OFFSET = None

  _plugin_classes = {}



  @classmethod
  def GetFormatSpecification(cls):
    """Retrieves the format specification.

    Returns:
      FormatSpecification: format specification.
    """
    format_specification = specification.FormatSpecification(cls.NAME)

    # OLECF
    format_specification.AddNewSignature(
        b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1', offset=0)

    # OLECF beta
    format_specification.AddNewSignature(
        b'\x0e\x11\xfc\x0d\xd0\xcf\x11\x0e', offset=0)

    return format_specification




  def ParseFileObject(self, parser_mediator, file_object):
    """Parses an OLE Compound File (OLECF) file-like object.

    Args:
      parser_mediator (ParserMediator): mediates interactions between parsers
          and other components, such as storage and dfVFS.
      file_object (dfvfs.FileIO): file-like object.
    """
    code_page = parser_mediator.GetCodePage()

    olecf_file = pyolecf.file()
    olecf_file.set_ascii_codepage(code_page)

    try:
      olecf_file.open_file_object(file_object)
    except (OSError, TypeError) as exception:
      parser_mediator.ProduceExtractionWarning(
          f'unable to open file with error: {exception!s}')
      return

    root_item = olecf_file.root_item
    if not root_item:
      return

    display_name = parser_mediator.GetDisplayName()

    # Get a list of all items in the root item from the OLECF file.
    item_names = [item.name for item in root_item.sub_items]

    # Compare the list of available plugin objects.
    # We will try to use every plugin against the file (except
    # the default plugin) and run it. Only if none of the plugins
    # works will we use the default plugin.

    item_names = frozenset(item_names)

    try:
      for plugin_name, plugin in self._plugins_per_name.items():
        if parser_mediator.abort:
          break

        profiling_name = '/'.join([self.NAME, plugin.NAME])

        parser_mediator.SampleFormatCheckStartTiming(profiling_name)

        try:
          result = plugin.REQUIRED_ITEMS.issubset(item_names)
        finally:
          parser_mediator.SampleFormatCheckStopTiming(profiling_name)

        if not result:
          logger.debug(
              f'Skipped parsing file: {display_name:s} with plugin: '
              f'{plugin_name:s}')
          continue

        logger.debug(
            f'Parsing file: {display_name:s} with plugin: {plugin_name:s}')

        parser_mediator.SampleStartTiming(profiling_name)

        try:
          plugin.UpdateChainAndProcess(parser_mediator, root_item=root_item)

        except Exception as exception:  # pylint: disable=broad-except
          parser_mediator.ProduceExtractionWarning(
              f'plugin: {plugin_name:s} unable to parse OLECF file with error: '
              f'{exception!s}')

        finally:
          parser_mediator.SampleStopTiming(profiling_name)

      if self._default_plugin and not parser_mediator.abort:
        profiling_name = '/'.join([self.NAME, self._default_plugin.NAME])

        parser_mediator.SampleStartTiming(profiling_name)

        try:
          self._default_plugin.UpdateChainAndProcess(
              parser_mediator, root_item=root_item)

        except Exception as exception:  # pylint: disable=broad-except
          parser_mediator.ProduceExtractionWarning(
              f'plugin: {self._default_plugin_name:s} unable to parse OLECF '
              f'file with error: {exception!s}')

        finally:
          parser_mediator.SampleStopTiming(profiling_name)

    finally:
      olecf_file.close()




manager.ParsersManager.RegisterParser(OLECFParser)