Source code for plaso.parsers.jsonl_plugins.aws_cloudtrail_log

"""JSON-L parser plugin for AWS CloudTrail log files."""

import json

from json import decoder as json_decoder

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 AWSCloudTrailEventData(events.EventData):
    """AWS CloudTrail log event data.

    Attributes:
      access_key (str): access key identifier.
      account_identifier (str): AWS account identifier.
      cloud_trail_event (str): CloudTrail event.
      event_name (str): event name.
      event_source (str): AWS service.
      recorded_time (dfdatetime.DateTimeValues): date and time the log entry was
          recorded.
      resources (str): resources.
      source_ip (str): source IP address.
      user_identity_arn (str): AWS ARN of the user.
      username (str): username.
    """

    DATA_TYPE = "aws:cloudtrail:entry"



    def __init__(self):
        """Initializes event data."""
        super().__init__(data_type=self.DATA_TYPE)
        self.access_key = None
        self.account_identifier = None
        self.cloud_trail_event = None
        self.event_name = None
        self.event_source = None
        self.recorded_time = None
        self.resources = None
        self.source_ip = None
        self.user_identity_arn = None
        self.username = None






class AWSCloudTrailLogJSONLPlugin(interface.JSONLPlugin):
    """JSON-L parser plugin for AWS CloudTrail log files."""

    NAME = "aws_cloudtrail_log"
    DATA_FORMAT = "AWS CloudTrail Log"

    def _ParseDateTimeString(self, parser_mediator, json_dict, name):
        """Parses a 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.TimeElementsInMicroseconds: date and time value or None if
                  not available.
              bool: value to indicate the date and time string was corrupted.
        """
        date_time_string = self._GetJSONValue(json_dict, name)
        if not date_time_string:
            return None, False

        try:
            date_time = dfdatetime_time_elements.TimeElementsInMicroseconds()
            date_time.CopyFromDateTimeString(date_time_string)
        except ValueError as exception:
            parser_mediator.ProduceWarning(
                f"Unable to parse value: {name:s} date time string: "
                f"{date_time_string:s} with error: {exception!s}"
            )
            return None, True

        return date_time, False

    def _ParseRecord(self, parser_mediator, json_dict):
        """Parses an AWS CloudTrail 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._ParseDateTimeString(
            parser_mediator, json_dict, "EventTime"
        )

        resource_list = self._GetJSONValue(json_dict, "Resources", default_value=[])
        # Flatten multiple resources into a string of resource names.
        resources = ", ".join([resource["ResourceName"] for resource in resource_list])

        cloud_trail_event = self._GetJSONValue(
            json_dict, "CloudTrailEvent", default_value="{}"
        )
        try:
            cloud_trail_event_json = json.loads(cloud_trail_event)
        except json_decoder.JSONDecodeError as exception:
            parser_mediator.ProduceWarning(
                f"Unable to decode CloudTrail event with error: {exception!s}"
            )
            cloud_trail_event_json = {}
            corrupted = True

        user_identity_json = self._GetJSONValue(cloud_trail_event_json, "userIdentity")

        event_data = AWSCloudTrailEventData()
        event_data.access_key = self._GetJSONValue(json_dict, "AccessKeyId")
        event_data.account_identifier = self._GetJSONValue(
            cloud_trail_event_json, "accountId"
        )
        event_data.cloud_trail_event = cloud_trail_event
        event_data.event_name = self._GetJSONValue(json_dict, "EventName")
        event_data.event_source = self._GetJSONValue(json_dict, "EventSource")
        event_data.recorded_time = date_time
        event_data.resources = resources or None
        event_data.source_ip = self._GetJSONValue(
            cloud_trail_event_json, "sourceIPAddress"
        )
        event_data.user_identity_arn = self._GetJSONValue(user_identity_json, "arn")
        event_data.username = self._GetJSONValue(json_dict, "Username")

        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 parser, False otherwise.
        """
        cloud_trail_event = self._GetJSONValue(json_dict, "CloudTrailEvent")
        event_time = self._GetJSONValue(json_dict, "EventTime")

        if None in (cloud_trail_event, event_time):
            return False

        date_time = dfdatetime_time_elements.TimeElementsInMicroseconds()

        try:
            date_time.CopyFromDateTimeString(event_time)
        except ValueError:
            return False

        return True




jsonl_parser.JSONLParser.RegisterPlugin(AWSCloudTrailLogJSONLPlugin)