Source code for dynamodb_encryption_sdk.internal.formatting.deserialize

# 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.
"""Helper functions for deserializing values.

.. warning::
    No guarantee is provided on the modules and APIs within this
    namespace staying consistent. Directly reference at your own risk.
"""
import struct

from dynamodb_encryption_sdk.exceptions import DeserializationError

__all__ = ("unpack_value", "decode_length", "decode_value", "decode_tag")


[docs]def unpack_value(format_string, stream): """Helper function to unpack struct data from a stream and update the signature verifier. :param str format_string: Struct format string :param stream: Source data stream :type stream: io.BytesIO :returns: Unpacked values :rtype: tuple """ message_bytes = stream.read(struct.calcsize(format_string)) return struct.unpack(format_string, message_bytes)
[docs]def decode_length(stream): """Decode the length of a value from a serialized stream. :param stream: Source data stream :type stream: io.BytesIO :returns: Decoded length :rtype: int """ (value,) = unpack_value(">I", stream) return value
[docs]def decode_value(stream): """Decode the contents of a value from a serialized stream. :param stream: Source data stream :type stream: io.BytesIO :returns: Decoded value :rtype: bytes """ length = decode_length(stream) (value,) = unpack_value(">{:d}s".format(length), stream) return value
def decode_byte(stream): """Decode a single raw byte from a serialized stream (used for deserialize bool). :param stream: Source data stream :type stream: io.BytesIO :returns: Decoded value :rtype: bytes """ (value,) = unpack_value(">1s", stream) return value
[docs]def decode_tag(stream): """Decode a tag value from a serialized stream. :param stream: Source data stream :type stream: io.BytesIO :returns: Decoded tag :rtype: bytes """ (reserved, tag) = unpack_value(">cc", stream) if reserved != b"\x00": raise DeserializationError("Invalid tag: reserved byte is not null") return tag