diff --git a/src/aws_encryption_sdk/internal/formatting/deserialize.py b/src/aws_encryption_sdk/internal/formatting/deserialize.py index c90dc9124..f8a6239b7 100644 --- a/src/aws_encryption_sdk/internal/formatting/deserialize.py +++ b/src/aws_encryption_sdk/internal/formatting/deserialize.py @@ -344,6 +344,9 @@ def deserialize_header(stream, max_encrypted_data_keys=None): tee = io.BytesIO() tee_stream = TeeStream(stream, tee) (version_id,) = unpack_values(">B", tee_stream) + # A base64-encoded message starts with 0x41 0x59 (V1: 0x01 0x80) or 0x41 0x67 (V2: 0x02 0x04/0x05). + if version_id == 0x41 and tee_stream.read(1) in (b"\x59", b"\x67"): + raise NotSupportedError("Unsupported version {}: message may be base64 encoded".format(version_id)) version = _verified_version_from_id(version_id) header = {} header["version"] = version diff --git a/test/unit/test_deserialize.py b/test/unit/test_deserialize.py index f0273c428..fec84f251 100644 --- a/test/unit/test_deserialize.py +++ b/test/unit/test_deserialize.py @@ -1,6 +1,7 @@ # Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 """Unit test suite for aws_encryption_sdk.deserialize""" +import base64 import io import struct @@ -129,6 +130,29 @@ def test_deserialize_header_unknown_version(self): stream = io.BytesIO(VALUES["serialized_header_invalid_version"]) aws_encryption_sdk.internal.formatting.deserialize.deserialize_header(stream) excinfo.match("Unsupported version *") + assert "base64" not in str(excinfo.value) + + @pytest.mark.parametrize( + "serialized_header", (VALUES["serialized_header"], VALUES["serialized_header_v2_committing"]) + ) + def test_deserialize_header_base64_encoded(self, serialized_header): + """Validate that the deserialize_header function points at base64 encoding + as the likely cause when handed a message that was not decoded first. + """ + with pytest.raises(NotSupportedError) as excinfo: + stream = io.BytesIO(base64.b64encode(serialized_header)) + aws_encryption_sdk.internal.formatting.deserialize.deserialize_header(stream) + excinfo.match("Unsupported version 65: message may be base64 encoded") + + @pytest.mark.parametrize("data", (b"A", b"AX", b"BAYA")) + def test_deserialize_header_without_base64_prefix(self, data): + """Validate that the deserialize_header function does not claim base64 encoding + unless both bytes of the expected base64 prefix match. + """ + with pytest.raises(NotSupportedError) as excinfo: + aws_encryption_sdk.internal.formatting.deserialize.deserialize_header(io.BytesIO(data)) + excinfo.match("Unsupported version *") + assert "base64" not in str(excinfo.value) @patch("aws_encryption_sdk.internal.formatting.deserialize.AlgorithmSuite.get_by_id") def test_deserialize_header_unsupported_data_encryption_algorithm(self, mock_algorithm_get):