AWS-DDNS/Function Source Files/botocore/eventstream.py

# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
"""Binary Event Stream Decoding """

from binascii import crc32
from struct import unpack

from botocore.exceptions import EventStreamError

# byte length of the prelude (total_length + header_length + prelude_crc)
_PRELUDE_LENGTH = 12
_MAX_HEADERS_LENGTH = 128 * 1024  # 128 Kb
_MAX_PAYLOAD_LENGTH = 16 * 1024**2  # 16 Mb


class ParserError(Exception):
    """Base binary flow encoding parsing exception."""

    pass


class DuplicateHeader(ParserError):
    """Duplicate header found in the event."""

    def __init__(self, header):
        message = 'Duplicate header present: "%s"' % header
        super().__init__(message)


class InvalidHeadersLength(ParserError):
    """Headers length is longer than the maximum."""

    def __init__(self, length):
        message = 'Header length of {} exceeded the maximum of {}'.format(
            length,
            _MAX_HEADERS_LENGTH,
        )
        super().__init__(message)


class InvalidPayloadLength(ParserError):
    """Payload length is longer than the maximum."""

    def __init__(self, length):
        message = 'Payload length of {} exceeded the maximum of {}'.format(
            length,
            _MAX_PAYLOAD_LENGTH,
        )
        super().__init__(message)


class ChecksumMismatch(ParserError):
    """Calculated checksum did not match the expected checksum."""

    def __init__(self, expected, calculated):
        message = (
            'Checksum mismatch: expected 0x{:08x}, calculated 0x{:08x}'.format(
                expected,
                calculated,
            )
        )
        super().__init__(message)


class NoInitialResponseError(ParserError):
    """An event of type initial-response was not received.

    This exception is raised when the event stream produced no events or
    the first event in the stream was not of the initial-response type.
    """

    def __init__(self):
        message = 'First event was not of the initial-response type'
        super().__init__(message)


class DecodeUtils:
    """Unpacking utility functions used in the decoder.

    All methods on this class take raw bytes and return  a tuple containing
    the value parsed from the bytes and the number of bytes consumed to parse
    that value.
    """

    UINT8_BYTE_FORMAT = '!B'
    UINT16_BYTE_FORMAT = '!H'
    UINT32_BYTE_FORMAT = '!I'
    INT8_BYTE_FORMAT = '!b'
    INT16_BYTE_FORMAT = '!h'
    INT32_BYTE_FORMAT = '!i'
    INT64_BYTE_FORMAT = '!q'
    PRELUDE_BYTE_FORMAT = '!III'

    # uint byte size to unpack format
    UINT_BYTE_FORMAT = {
        1: UINT8_BYTE_FORMAT,
        2: UINT16_BYTE_FORMAT,
        4: UINT32_BYTE_FORMAT,
    }

    @staticmethod
    def unpack_true(data):
        """This method consumes none of the provided bytes and returns True.

        :type data: bytes
        :param data: The bytes to parse from. This is ignored in this method.

        :rtype: tuple
        :rtype: (bool, int)
        :returns: The tuple (True, 0)
        """
        return True, 0

    @staticmethod
    def unpack_false(data):
        """This method consumes none of the provided bytes and returns False.

        :type data: bytes
        :param data: The bytes to parse from. This is ignored in this method.

        :rtype: tuple
        :rtype: (bool, int)
        :returns: The tuple (False, 0)
        """
        return False, 0

    @staticmethod
    def unpack_uint8(data):
        """Parse an unsigned 8-bit integer from the bytes.

        :type data: bytes
        :param data: The bytes to parse from.

        :rtype: (int, int)
        :returns: A tuple containing the (parsed integer value, bytes consumed)
        """
        value = unpack(DecodeUtils.UINT8_BYTE_FORMAT, data[:1])[0]
        return value, 1

    @staticmethod
    def unpack_uint32(data):
        """Parse an unsigned 32-bit integer from the bytes.

        :type data: bytes
        :param data: The bytes to parse from.

        :rtype: (int, int)
        :returns: A tuple containing the (parsed integer value, bytes consumed)
        """
        value = unpack(DecodeUtils.UINT32_BYTE_FORMAT, data[:4])[0]
        return value, 4

    @staticmethod
    def unpack_int8(data):
        """Parse a signed 8-bit integer from the bytes.

        :type data: bytes
        :param data: The bytes to parse from.

        :rtype: (int, int)
        :returns: A tuple containing the (parsed integer value, bytes consumed)
        """
        value = unpack(DecodeUtils.INT8_BYTE_FORMAT, data[:1])[0]
        return value, 1

    @staticmethod
    def unpack_int16(data):
        """Parse a signed 16-bit integer from the bytes.

        :type data: bytes
        :param data: The bytes to parse from.

        :rtype: tuple
        :rtype: (int, int)
        :returns: A tuple containing the (parsed integer value, bytes consumed)
        """
        value = unpack(DecodeUtils.INT16_BYTE_FORMAT, data[:2])[0]
        return value, 2

    @staticmethod
    def unpack_int32(data):
        """Parse a signed 32-bit integer from the bytes.

        :type data: bytes
        :param data: The bytes to parse from.

        :rtype: tuple
        :rtype: (int, int)
        :returns: A tuple containing the (parsed integer value, bytes consumed)
        """
        value = unpack(DecodeUtils.INT32_BYTE_FORMAT, data[:4])[0]
        return value, 4

    @staticmethod
    def unpack_int64(data):
        """Parse a signed 64-bit integer from the bytes.

        :type data: bytes
        :param data: The bytes to parse from.

        :rtype: tuple
        :rtype: (int, int)
        :returns: A tuple containing the (parsed integer value, bytes consumed)
        """
        value = unpack(DecodeUtils.INT64_BYTE_FORMAT, data[:8])[0]
        return value, 8

    @staticmethod
    def unpack_byte_array(data, length_byte_size=2):
        """Parse a variable length byte array from the bytes.

        The bytes are expected to be in the following format:
            [ length ][0 ... length bytes]
        where length is an unsigned integer represented in the smallest number
        of bytes to hold the maximum length of the array.

        :type data: bytes
        :param data: The bytes to parse from.

        :type length_byte_size: int
        :param length_byte_size: The byte size of the preceding integer that
        represents the length of the array. Supported values are 1, 2, and 4.

        :rtype: (bytes, int)
        :returns: A tuple containing the (parsed byte array, bytes consumed).
        """
        uint_byte_format = DecodeUtils.UINT_BYTE_FORMAT[length_byte_size]
        length = unpack(uint_byte_format, data[:length_byte_size])[0]
        bytes_end = length + length_byte_size
        array_bytes = data[length_byte_size:bytes_end]
        return array_bytes, bytes_end

    @staticmethod
    def unpack_utf8_string(data, length_byte_size=2):
        """Parse a variable length utf-8 string from the bytes.

        The bytes are expected to be in the following format:
            [ length ][0 ... length bytes]
        where length is an unsigned integer represented in the smallest number
        of bytes to hold the maximum length of the array and the following
        bytes are a valid utf-8 string.

        :type data: bytes
        :param bytes: The bytes to parse from.

        :type length_byte_size: int
        :param length_byte_size: The byte size of the preceding integer that
        represents the length of the array. Supported values are 1, 2, and 4.

        :rtype: (str, int)
        :returns: A tuple containing the (utf-8 string, bytes consumed).
        """
        array_bytes, consumed = DecodeUtils.unpack_byte_array(
            data, length_byte_size
        )
        return array_bytes.decode('utf-8'), consumed

    @staticmethod
    def unpack_uuid(data):
        """Parse a 16-byte uuid from the bytes.

        :type data: bytes
        :param data: The bytes to parse from.

        :rtype: (bytes, int)
        :returns: A tuple containing the (uuid bytes, bytes consumed).
        """
        return data[:16], 16

    @staticmethod
    def unpack_prelude(data):
        """Parse the prelude for an event stream message from the bytes.

        The prelude for an event stream message has the following format:
            [total_length][header_length][prelude_crc]
        where each field is an unsigned 32-bit integer.

        :rtype: ((int, int, int), int)
        :returns: A tuple of ((total_length, headers_length, prelude_crc),
        consumed)
        """
        return (unpack(DecodeUtils.PRELUDE_BYTE_FORMAT, data), _PRELUDE_LENGTH)


def _validate_checksum(data, checksum, crc=0):
    # To generate the same numeric value across all Python versions and
    # platforms use crc32(data) & 0xffffffff.
    computed_checksum = crc32(data, crc) & 0xFFFFFFFF
    if checksum != computed_checksum:
        raise ChecksumMismatch(checksum, computed_checksum)


class MessagePrelude:
    """Represents the prelude of an event stream message."""

    def __init__(self, total_length, headers_length, crc):
        self.total_length = total_length
        self.headers_length = headers_length
        self.crc = crc

    @property
    def payload_length(self):
        """Calculates the total payload length.

        The extra minus 4 bytes is for the message CRC.

        :rtype: int
        :returns: The total payload length.
        """
        return self.total_length - self.headers_length - _PRELUDE_LENGTH - 4

    @property
    def payload_end(self):
        """Calculates the byte offset for the end of the message payload.

        The extra minus 4 bytes is for the message CRC.

        :rtype: int
        :returns: The byte offset from the beginning of the event stream
        message to the end of the payload.
        """
        return self.total_length - 4

    @property
    def headers_end(self):
        """Calculates the byte offset for the end of the message headers.

        :rtype: int
        :returns: The byte offset from the beginning of the event stream
        message to the end of the headers.
        """
        return _PRELUDE_LENGTH + self.headers_length


class EventStreamMessage:
    """Represents an event stream message."""

    def __init__(self, prelude, headers, payload, crc):
        self.prelude = prelude
        self.headers = headers
        self.payload = payload
        self.crc = crc

    def to_response_dict(self, status_code=200):
        message_type = self.headers.get(':message-type')
        if message_type == 'error' or message_type == 'exception':
            status_code = 400
        return {
            'status_code': status_code,
            'headers': self.headers,
            'body': self.payload,
        }


class EventStreamHeaderParser:
    """Parses the event headers from an event stream message.

    Expects all of the header data upfront and creates a dictionary of headers
    to return. This object can be reused multiple times to parse the headers
    from multiple event stream messages.
    """

    # Maps header type to appropriate unpacking function
    # These unpacking functions return the value and the amount unpacked
    _HEADER_TYPE_MAP = {
        # boolean_true
        0: DecodeUtils.unpack_true,
        # boolean_false
        1: DecodeUtils.unpack_false,
        # byte
        2: DecodeUtils.unpack_int8,
        # short
        3: DecodeUtils.unpack_int16,
        # integer
        4: DecodeUtils.unpack_int32,
        # long
        5: DecodeUtils.unpack_int64,
        # byte_array
        6: DecodeUtils.unpack_byte_array,
        # string
        7: DecodeUtils.unpack_utf8_string,
        # timestamp
        8: DecodeUtils.unpack_int64,
        # uuid
        9: DecodeUtils.unpack_uuid,
    }

    def __init__(self):
        self._data = None

    def parse(self, data):
        """Parses the event stream headers from an event stream message.

        :type data: bytes
        :param data: The bytes that correspond to the headers section of an
        event stream message.

        :rtype: dict
        :returns: A dictionary of header key, value pairs.
        """
        self._data = data
        return self._parse_headers()

    def _parse_headers(self):
        headers = {}
        while self._data:
            name, value = self._parse_header()
            if name in headers:
                raise DuplicateHeader(name)
            headers[name] = value
        return headers

    def _parse_header(self):
        name = self._parse_name()
        value = self._parse_value()
        return name, value

    def _parse_name(self):
        name, consumed = DecodeUtils.unpack_utf8_string(self._data, 1)
        self._advance_data(consumed)
        return name

    def _parse_type(self):
        type, consumed = DecodeUtils.unpack_uint8(self._data)
        self._advance_data(consumed)
        return type

    def _parse_value(self):
        header_type = self._parse_type()
        value_unpacker = self._HEADER_TYPE_MAP[header_type]
        value, consumed = value_unpacker(self._data)
        self._advance_data(consumed)
        return value

    def _advance_data(self, consumed):
        self._data = self._data[consumed:]


class EventStreamBuffer:
    """Streaming based event stream buffer

    A buffer class that wraps bytes from an event stream providing parsed
    messages as they become available via an iterable interface.
    """

    def __init__(self):
        self._data = b''
        self._prelude = None
        self._header_parser = EventStreamHeaderParser()

    def add_data(self, data):
        """Add data to the buffer.

        :type data: bytes
        :param data: The bytes to add to the buffer to be used when parsing
        """
        self._data += data

    def _validate_prelude(self, prelude):
        if prelude.headers_length > _MAX_HEADERS_LENGTH:
            raise InvalidHeadersLength(prelude.headers_length)

        if prelude.payload_length > _MAX_PAYLOAD_LENGTH:
            raise InvalidPayloadLength(prelude.payload_length)

    def _parse_prelude(self):
        prelude_bytes = self._data[:_PRELUDE_LENGTH]
        raw_prelude, _ = DecodeUtils.unpack_prelude(prelude_bytes)
        prelude = MessagePrelude(*raw_prelude)
        self._validate_prelude(prelude)
        # The minus 4 removes the prelude crc from the bytes to be checked
        _validate_checksum(prelude_bytes[: _PRELUDE_LENGTH - 4], prelude.crc)
        return prelude

    def _parse_headers(self):
        header_bytes = self._data[_PRELUDE_LENGTH : self._prelude.headers_end]
        return self._header_parser.parse(header_bytes)

    def _parse_payload(self):
        prelude = self._prelude
        payload_bytes = self._data[prelude.headers_end : prelude.payload_end]
        return payload_bytes

    def _parse_message_crc(self):
        prelude = self._prelude
        crc_bytes = self._data[prelude.payload_end : prelude.total_length]
        message_crc, _ = DecodeUtils.unpack_uint32(crc_bytes)
        return message_crc

    def _parse_message_bytes(self):
        # The minus 4 includes the prelude crc to the bytes to be checked
        message_bytes = self._data[
            _PRELUDE_LENGTH - 4 : self._prelude.payload_end
        ]
        return message_bytes

    def _validate_message_crc(self):
        message_crc = self._parse_message_crc()
        message_bytes = self._parse_message_bytes()
        _validate_checksum(message_bytes, message_crc, crc=self._prelude.crc)
        return message_crc

    def _parse_message(self):
        crc = self._validate_message_crc()
        headers = self._parse_headers()
        payload = self._parse_payload()
        message = EventStreamMessage(self._prelude, headers, payload, crc)
        self._prepare_for_next_message()
        return message

    def _prepare_for_next_message(self):
        # Advance the data and reset the current prelude
        self._data = self._data[self._prelude.total_length :]
        self._prelude = None

    def next(self):
        """Provides the next available message parsed from the stream

        :rtype: EventStreamMessage
        :returns: The next event stream message
        """
        if len(self._data) < _PRELUDE_LENGTH:
            raise StopIteration()

        if self._prelude is None:
            self._prelude = self._parse_prelude()

        if len(self._data) < self._prelude.total_length:
            raise StopIteration()

        return self._parse_message()

    def __next__(self):
        return self.next()

    def __iter__(self):
        return self


class EventStream:
    """Wrapper class for an event stream body.

    This wraps the underlying streaming body, parsing it for individual events
    and yielding them as they come available through the iterator interface.

    The following example uses the S3 select API to get structured data out of
    an object stored in S3 using an event stream.

    **Example:**
    ::
        from botocore.session import Session

        s3 = Session().create_client('s3')
        response = s3.select_object_content(
            Bucket='bucketname',
            Key='keyname',
            ExpressionType='SQL',
            RequestProgress={'Enabled': True},
            Expression="SELECT * FROM S3Object s",
            InputSerialization={'CSV': {}},
            OutputSerialization={'CSV': {}},
        )
        # This is the event stream in the response
        event_stream = response['Payload']
        end_event_received = False
        with open('output', 'wb') as f:
            # Iterate over events in the event stream as they come
            for event in event_stream:
                # If we received a records event, write the data to a file
                if 'Records' in event:
                    data = event['Records']['Payload']
                    f.write(data)
                # If we received a progress event, print the details
                elif 'Progress' in event:
                    print(event['Progress']['Details'])
                # End event indicates that the request finished successfully
                elif 'End' in event:
                    print('Result is complete')
                    end_event_received = True
        if not end_event_received:
            raise Exception("End event not received, request incomplete.")
    """

    def __init__(self, raw_stream, output_shape, parser, operation_name):
        self._raw_stream = raw_stream
        self._output_shape = output_shape
        self._operation_name = operation_name
        self._parser = parser
        self._event_generator = self._create_raw_event_generator()

    def __iter__(self):
        for event in self._event_generator:
            parsed_event = self._parse_event(event)
            if parsed_event:
                yield parsed_event

    def _create_raw_event_generator(self):
        event_stream_buffer = EventStreamBuffer()
        for chunk in self._raw_stream.stream():
            event_stream_buffer.add_data(chunk)
            yield from event_stream_buffer

    def _parse_event(self, event):
        response_dict = event.to_response_dict()
        parsed_response = self._parser.parse(response_dict, self._output_shape)
        if response_dict['status_code'] == 200:
            return parsed_response
        else:
            raise EventStreamError(parsed_response, self._operation_name)

    def get_initial_response(self):
        try:
            initial_event = next(self._event_generator)
            event_type = initial_event.headers.get(':event-type')
            if event_type == 'initial-response':
                return initial_event
        except StopIteration:
            pass
        raise NoInitialResponseError()

    def close(self):
        """Closes the underlying streaming body."""
        self._raw_stream.close()