[SchemaRegistry] fix avro type leaks in serializer

sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/_avro_serializer.py

@@ -26,7 +26,9 @@
 from typing import BinaryIO, Union, TypeVar, Dict
 from io import BytesIO
 import avro
-from avro.io import DatumWriter, DatumReader, BinaryDecoder, BinaryEncoder
+from avro.schema import SchemaParseException, AvroException, InvalidName
+from avro.io import DatumWriter, DatumReader, BinaryDecoder, BinaryEncoder, AvroTypeException
+from .exceptions import SchemaParseException as AzureSchemaParseException
 
 ObjectType = TypeVar("ObjectType")
 
@@ -43,25 +45,28 @@ def __init__(self, codec=None):
 
     def serialize(
         self,
-        data,  # type: ObjectType
-        schema,  # type: Union[str, bytes, avro.schema.Schema]
+        data,  # type: Mapping[str, Any]
+        schema,  # type: Union[str, avro.schema.Schema]
     ):
         # type: (ObjectType, Union[str, bytes, avro.schema.Schema]) -> bytes
         """Convert the provided value to it's binary representation and write it to the stream.
         Schema must be a Avro RecordSchema:
         https://avro.apache.org/docs/1.10.0/gettingstartedpython.html#Defining+a+schema
-        :param data: An object to serialize
-        :type data: ObjectType
+        :param data: A data to serialize
+        :type data: Mapping[str, Any]
         :param schema: An Avro RecordSchema
-        :type schema: Union[str, bytes, avro.schema.Schema]
+        :type schema: Union[str, avro.schema.Schema]
         :returns: Encoded bytes
         :rtype: bytes
         """
         if not schema:
             raise ValueError("Schema is required in Avro serializer.")
 
         if not isinstance(schema, avro.schema.Schema):
-            schema = avro.schema.parse(schema)
+            try:
+                schema = avro.schema.parse(schema)
+            except (SchemaParseException, AvroTypeException, AvroException, InvalidName) as e:
+                raise AzureSchemaParseException(e)
 
         try:
             writer = self._schema_writer_cache[str(schema)]
@@ -71,30 +76,36 @@ def serialize(
 
         stream = BytesIO()
         with stream:
-            writer.write(data, BinaryEncoder(stream))
+            try:
+                writer.write(data, BinaryEncoder(stream))
+            except AvroTypeException as e:
+                raise ValueError(e)
             encoded_data = stream.getvalue()
         return encoded_data
 
     def deserialize(
         self,
         data,  # type: Union[bytes, BinaryIO]
-        schema,  # type:  Union[str, bytes, avro.schema.Schema]
+        schema,  # type:  str
     ):
-        # type: (Union[bytes, BinaryIO], Union[str, bytes, avro.schema.Schema]) -> ObjectType
+        # type: (Union[bytes, BinaryIO], str) -> ObjectType
         """Read the binary representation into a specific type.
         Return type will be ignored, since the schema is deduced from the provided bytes.
         :param data: A stream of bytes or bytes directly
         :type data: BinaryIO or bytes
         :param schema: An Avro RecordSchema
-        :type schema: Union[str, bytes, avro.schema.Schema]
+        :type schema: str
         :returns: An instantiated object
         :rtype: ObjectType
         """
         if not hasattr(data, 'read'):
             data = BytesIO(data)
 
         if not isinstance(schema, avro.schema.Schema):
-            schema = avro.schema.parse(schema)
+            try:
+                schema = avro.schema.parse(schema)
+            except (SchemaParseException, AvroTypeException, AvroException, InvalidName) as e:
+                raise AzureSchemaParseException(e)
 
         try:
             reader = self._schema_reader_cache[str(schema)]
@@ -104,6 +115,9 @@ def deserialize(
 
         with data:
             bin_decoder = BinaryDecoder(data)
-            decoded_data = reader.read(bin_decoder)
+            try:
+                decoded_data = reader.read(bin_decoder)
+            except AvroTypeException as e:
+                raise ValueError(e)
 
         return decoded_data

sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/_schema_registry_avro_serializer.py

@@ -30,7 +30,15 @@
 from io import BytesIO
 from typing import Any, Dict, Mapping
 import avro
-
+from avro.schema import (
+    SchemaParseException,
+    AvroException,
+    InvalidName,
+    PRIMITIVE_TYPES,
+)
+from avro.io import AvroTypeException
+
+from .exceptions import SchemaParseException as AzureSchemaParseException
 from ._constants import SCHEMA_ID_START_INDEX, SCHEMA_ID_LENGTH, DATA_START_INDEX
 from ._avro_serializer import AvroObjectSerializer
 
@@ -53,16 +61,18 @@ def __init__(self, **kwargs):
         # type: (Any) -> None
         try:
             self._schema_group = kwargs.pop("group_name")
-            self._schema_registry_client = kwargs.pop("client") # type: "SchemaRegistryClient"
+            self._schema_registry_client = kwargs.pop(
+                "client"
+            )  # type: "SchemaRegistryClient"
         except KeyError as e:
             raise TypeError("'{}' is a required keyword.".format(e.args[0]))
         self._avro_serializer = AvroObjectSerializer(codec=kwargs.get("codec"))
         self._auto_register_schemas = kwargs.get("auto_register_schemas", False)
         self._auto_register_schema_func = (
-                self._schema_registry_client.register_schema
-                if self._auto_register_schemas
-                else self._schema_registry_client.get_schema_id
-            )
+            self._schema_registry_client.register_schema
+            if self._auto_register_schemas
+            else self._schema_registry_client.get_schema_id
+        )
         self._user_input_schema_cache = {}
 
     def __enter__(self):
@@ -121,10 +131,12 @@ def serialize(self, value, **kwargs):
         Encode data with the given schema. The returns bytes are consisted of: The first 4 bytes
         denoting record format identifier. The following 32 bytes denoting schema id returned by schema registry
         service. The remaining bytes are the real data payload.
+        Schema must be a Avro RecordSchema:
+        https://avro.apache.org/docs/1.10.0/gettingstartedpython.html#Defining+a+schema
 
         :param value: The data to be encoded.
         :type value: Mapping[str, Any]
-        :keyword schema: Required. The schema used to encode the data.
+        :keyword schema: Required. The Avro RecordSchema used to encode the data.
         :paramtype schema: str
         :rtype: bytes
         """
@@ -135,12 +147,37 @@ def serialize(self, value, **kwargs):
         try:
             cached_schema = self._user_input_schema_cache[raw_input_schema]
         except KeyError:
-            parsed_schema = avro.schema.parse(raw_input_schema)
+            try:
+                parsed_schema = avro.schema.parse(raw_input_schema)
+            except (
+                SchemaParseException,
+                AvroTypeException,
+                AvroException,
+                InvalidName,
+            ) as e:
+                raise AzureSchemaParseException(e)
             self._user_input_schema_cache[raw_input_schema] = parsed_schema
             cached_schema = parsed_schema
 
+        # TODO: check if there's a more pythonic way, if we need it at all
+        if cached_schema.type in PRIMITIVE_TYPES:
+            raise ValueError(
+                "Expected schema of type `Record`, got instead primitive: {}".format(
+                    raw_input_schema
+                )
+            )
+
         record_format_identifier = b"\0\0\0\0"
-        schema_id = self._get_schema_id(cached_schema.fullname, str(cached_schema), **kwargs)
+        try:
+            schema_id = self._get_schema_id(
+                cached_schema.fullname, str(cached_schema), **kwargs
+            )
+        except AttributeError:
+            raise ValueError(
+                "Expected schema of type `Record`, got instead schema of type: {}".format(
+                    cached_schema.type
+                )
+            )
         data_bytes = self._avro_serializer.serialize(value, cached_schema)
 
         stream = BytesIO()

sdk/schemaregistry/azure-schemaregistry-avroserializer/azure/schemaregistry/serializer/avroserializer/exceptions.py

@@ -0,0 +1,40 @@
+# --------------------------------------------------------------------------
+#
+# Copyright (c) Microsoft Corporation. All rights reserved.
+#
+# The MIT License (MIT)
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the ""Software""), to
+# deal in the Software without restriction, including without limitation the
+# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+# sell copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+# IN THE SOFTWARE.
+#
+# --------------------------------------------------------------------------
+from azure.core.exceptions import AzureError
+
+class SchemaParseException(AzureError):
+    """Error parsing a JSON schema.
+
+    :ivar message: The error message.
+    :vartype message: str
+    :ivar error: The error condition, if available.
+    :vartype error: str
+    """
+
+    def __init__(self, message, **kwargs):
+        self.message = message
+        self.error = kwargs.get("error")
+        super(SchemaParseException, self).__init__(self.message, **kwargs)

sdk/schemaregistry/azure-schemaregistry-avroserializer/tests/recordings/test_avro_serializer.test_basic_sr_avro_serializer_with_auto_register_schemas.yaml

@@ -23,12 +23,12 @@ interactions:
     uri: https://fake_resource.servicebus.windows.net/$schemagroups/fakegroup/schemas/example.avro.User?api-version=2017-04
   response:
     body:
-      string: '{"id":"f666e373299048fabaa4296f5dbfed46"}'
+      string: '{"id":"7b4eff1c25d9438a975ff7a3d985a5c6"}'
     headers:
       content-type:
       - application/json
       date:
-      - Tue, 28 Sep 2021 22:27:25 GMT
+      - Fri, 01 Oct 2021 08:50:24 GMT
       location:
       - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/fakegroup/schemas/example.avro.User/versions/1?api-version=2017-04
       server:
@@ -38,9 +38,9 @@ interactions:
       transfer-encoding:
       - chunked
       x-schema-id:
-      - f666e373299048fabaa4296f5dbfed46
+      - 7b4eff1c25d9438a975ff7a3d985a5c6
       x-schema-id-location:
-      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/f666e373299048fabaa4296f5dbfed46?api-version=2017-04
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/7b4eff1c25d9438a975ff7a3d985a5c6?api-version=2017-04
       x-schema-type:
       - Avro
       x-schema-version:

sdk/schemaregistry/azure-schemaregistry-avroserializer/tests/recordings/test_avro_serializer.test_basic_sr_avro_serializer_without_auto_register_schemas.yaml

@@ -23,12 +23,12 @@ interactions:
     uri: https://fake_resource.servicebus.windows.net/$schemagroups/fakegroup/schemas/example.avro.User?api-version=2017-04
   response:
     body:
-      string: '{"id":"f666e373299048fabaa4296f5dbfed46"}'
+      string: '{"id":"7b4eff1c25d9438a975ff7a3d985a5c6"}'
     headers:
       content-type:
       - application/json
       date:
-      - Tue, 28 Sep 2021 22:27:26 GMT
+      - Fri, 01 Oct 2021 08:50:25 GMT
       location:
       - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/fakegroup/schemas/example.avro.User/versions/1?api-version=2017-04
       server:
@@ -38,9 +38,9 @@ interactions:
       transfer-encoding:
       - chunked
       x-schema-id:
-      - f666e373299048fabaa4296f5dbfed46
+      - 7b4eff1c25d9438a975ff7a3d985a5c6
       x-schema-id-location:
-      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/f666e373299048fabaa4296f5dbfed46?api-version=2017-04
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/7b4eff1c25d9438a975ff7a3d985a5c6?api-version=2017-04
       x-schema-type:
       - Avro
       x-schema-version:

sdk/schemaregistry/azure-schemaregistry-avroserializer/tests/recordings/test_avro_serializer.test_parse_error_schema_as_record.yaml

@@ -0,0 +1,51 @@
+interactions:
+- request:
+    body: '"{\"type\": \"error\", \"name\": \"User\", \"namespace\": \"example.avro.error\",
+      \"fields\": [{\"type\": \"string\", \"name\": \"name\"}]}"'
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '140'
+      Content-Type:
+      - application/json
+      User-Agent:
+      - azsdk-python-azureschemaregistry/1.0.0b1 Python/3.9.0 (Windows-10-10.0.19041-SP0)
+      X-Schema-Type:
+      - Avro
+    method: PUT
+    uri: https://fake_resource.servicebus.windows.net/$schemagroups/fakegroup/schemas/example.avro.error.User?api-version=2017-04
+  response:
+    body:
+      string: '{"id":"4e9f5888b6f8411abaa4ce6c110b2fb3"}'
+    headers:
+      content-type:
+      - application/json
+      date:
+      - Fri, 01 Oct 2021 08:50:26 GMT
+      location:
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/fakegroup/schemas/example.avro.error.User/versions/1?api-version=2017-04
+      server:
+      - Microsoft-HTTPAPI/2.0
+      strict-transport-security:
+      - max-age=31536000
+      transfer-encoding:
+      - chunked
+      x-schema-id:
+      - 4e9f5888b6f8411abaa4ce6c110b2fb3
+      x-schema-id-location:
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/4e9f5888b6f8411abaa4ce6c110b2fb3?api-version=2017-04
+      x-schema-type:
+      - Avro
+      x-schema-version:
+      - '1'
+      x-schema-versions-location:
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/swathip-test-schema/schemas/example.avro.error.User/versions?api-version=2017-04
+    status:
+      code: 200
+      message: OK
+version: 1

sdk/schemaregistry/azure-schemaregistry-avroserializer/tests/recordings/test_avro_serializer.test_parse_invalid_aliases.yaml

@@ -0,0 +1,52 @@
+interactions:
+- request:
+    body: '"{\"type\": \"record\", \"aliases\": [\"Resu\"], \"name\": \"User_aliases\",
+      \"namespace\": \"example.avro\", \"fields\": [{\"type\": \"string\", \"name\":
+      \"name\"}]}"'
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '168'
+      Content-Type:
+      - application/json
+      User-Agent:
+      - azsdk-python-azureschemaregistry/1.0.0b1 Python/3.9.0 (Windows-10-10.0.19041-SP0)
+      X-Schema-Type:
+      - Avro
+    method: PUT
+    uri: https://fake_resource.servicebus.windows.net/$schemagroups/fakegroup/schemas/example.avro.User_aliases?api-version=2017-04
+  response:
+    body:
+      string: '{"id":"83a1c5b1e3f049acae089ed34eb5191c"}'
+    headers:
+      content-type:
+      - application/json
+      date:
+      - Fri, 01 Oct 2021 07:29:00 GMT
+      location:
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/fakegroup/schemas/example.avro.User_aliases/versions/1?api-version=2017-04
+      server:
+      - Microsoft-HTTPAPI/2.0
+      strict-transport-security:
+      - max-age=31536000
+      transfer-encoding:
+      - chunked
+      x-schema-id:
+      - 83a1c5b1e3f049acae089ed34eb5191c
+      x-schema-id-location:
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/getschemabyid/83a1c5b1e3f049acae089ed34eb5191c?api-version=2017-04
+      x-schema-type:
+      - Avro
+      x-schema-version:
+      - '1'
+      x-schema-versions-location:
+      - https://swathip-test-eventhubs.servicebus.windows.net:443/$schemagroups/swathip-test-schema/schemas/example.avro.User_aliases/versions?api-version=2017-04
+    status:
+      code: 200
+      message: OK
+version: 1