Source code for plaso.parsers.jsonl_plugins.microsoft365_audit_log

"""JSON-L parser plugin for Microsoft (Office) 365 audit log files."""

from dfdatetime import time_elements as dfdatetime_time_elements

from plaso.containers import events
from plaso.parsers import jsonl_parser
from plaso.parsers.jsonl_plugins import interface




class Microsoft365AuditLogEventData(events.EventData):
    """Microsoft (Office) 365 audit log event data.

    Attributes:
      audit_record_identifier (str): audit record identifier.
      application_access_context (str): application access context
      client_ip (str): client IP address.
      object_identifier (str): object identifier
      operation_name (str): operation name.
      organization_identifier (str): organization identifier.
      record_type (int): record type.
      recorded_time (dfdatetime.DateTimeValues): date and time the log entry was
          recorded.
      result_status (str): result status
      scope (str): scope.
      user_identifier (str): user identifier
      user_key (str): user key.
      user_type (int): user type.
      workload (str): Microsoft (Office) 365 service
    """

    DATA_TYPE = "microsoft365:audit_log:entry"



    def __init__(self):
        """Initializes event data."""
        super().__init__(data_type=self.DATA_TYPE)
        self.audit_record_identifier = None
        self.application_access_context = None
        self.client_ip = None
        self.object_identifier = None
        self.operation_name = None
        self.organization_identifier = None
        self.record_type = None
        self.recorded_time = None
        self.result_status = None
        self.scope = None
        self.user_identifier = None
        self.user_key = None
        self.user_type = None
        self.workload = None






class Microsoft365AuditLogJSONLPlugin(interface.JSONLPlugin):
    """JSON-L parser plugin for Microsoft (Office) 365 audit log files."""

    NAME = "microsoft_audit_log"
    DATA_FORMAT = "Microsoft (Office) 365 audit log"

    def _ParseISO8601DateTimeString(self, parser_mediator, json_dict, name):
        """Parses an ISO8601 date and time string.

        Args:
          parser_mediator (ParserMediator): mediates interactions between parsers and
              other components, such as storage and dfVFS.
          json_dict (dict): JSON dictionary.
          name (str): name of the value to retrieve.

        Returns:
          tuple: containing:

              dfdatetime.TimeElements: date and time value or None if not available.
              bool: value to indicate the data and time string was corrupted.
        """
        iso8601_string = self._GetJSONValue(json_dict, name)
        if not iso8601_string:
            return None, False

        try:
            # Note that Microsoft (Office) 365 audit log uses seconds precision.
            date_time = dfdatetime_time_elements.TimeElements()
            date_time.CopyFromStringISO8601(iso8601_string)
        except ValueError as exception:
            parser_mediator.ProduceWarning(
                f"Unable to parse value: {name:s} ISO8601 string: {iso8601_string:s} "
                f"with error: {exception!s}"
            )
            return None, True

        return date_time, False

    def _ParseRecord(self, parser_mediator, json_dict):
        """Parses a Microsoft (Office) 365 audit log record.

        Args:
          parser_mediator (ParserMediator): mediates interactions between parsers and
              other components, such as storage and dfVFS.
          json_dict (dict): JSON dictionary of the log record.
        """
        date_time, corrupted = self._ParseISO8601DateTimeString(
            parser_mediator, json_dict, "CreationTime"
        )
        event_data = Microsoft365AuditLogEventData()

        event_data.audit_record_identifier = self._GetJSONValue(json_dict, "Id")
        event_data.application_access_context = self._GetJSONValue(
            json_dict, "AppAccessContext"
        )
        event_data.client_ip = self._GetJSONValue(json_dict, "ClientIP")
        event_data.object_identifier = self._GetJSONValue(json_dict, "ObjectId")
        event_data.operation_name = self._GetJSONValue(json_dict, "Operation")
        event_data.organization_identifier = self._GetJSONValue(
            json_dict, "OrganizationId"
        )
        event_data.record_type = self._GetJSONValue(json_dict, "RecordType")
        event_data.recorded_time = date_time
        event_data.result_status = self._GetJSONValue(json_dict, "ResultStatus")
        event_data.scope = self._GetJSONValue(json_dict, "Scope")
        event_data.user_identifier = self._GetJSONValue(json_dict, "UserId")
        event_data.user_key = self._GetJSONValue(json_dict, "UserKey")
        event_data.user_type = self._GetJSONValue(json_dict, "UserType")
        event_data.workload = self._GetJSONValue(json_dict, "Workload")

        parser_mediator.ProduceEventData(event_data, corrupted=corrupted)



    def CheckRequiredFormat(self, json_dict):
        """Check if the log record has the minimal structure required by the plugin.

        Args:
          json_dict (dict): JSON dictionary of the log record

        Returns:
          bool: True if this is the correct parsers, False otherwise.
        """
        audit_record_identifier = self._GetJSONValue(json_dict, "Id")
        organization_identifier = self._GetJSONValue(json_dict, "OrganizationId")
        creation_time = self._GetJSONValue(json_dict, "CreationTime")

        if None in (audit_record_identifier, creation_time, organization_identifier):
            return False

        date_time = dfdatetime_time_elements.TimeElements()

        try:
            date_time.CopyFromStringISO8601(creation_time)
        except ValueError:
            return False

        return True




jsonl_parser.JSONLParser.RegisterPlugin(Microsoft365AuditLogJSONLPlugin)