fix: hint that ciphertext may be base64 encoded

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

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):