raise IncompleteReadError if only receive partial response

sdk/core/azure-core/CHANGELOG.md

@@ -4,10 +4,15 @@
 
 ### Features Added
 
+- Added new error type `IncompleteReadError` which is raised if peer closes connection without sending complete message body.
+
 ### Breaking Changes
 
 ### Bugs Fixed
 
+- The `Content-Length` header in a http response is strictly checked against the actual number of bytes in the body,
+  rather than silently truncating data in case the underlying tcp connection is closed prematurely. 
+  (thanks to @jochen-ott-by for the contribution)   #20412
 - respect text encoding specified in argument (thanks to @ryohji for the contribution)  #20796
 
 ### Other Changes

sdk/core/azure-core/azure/core/exceptions.py

@@ -338,6 +338,10 @@ class DecodeError(HttpResponseError):
     """Error raised during response deserialization."""
 
 
+class IncompleteReadError(DecodeError):
+    """Error raised if peer closes connection without sending complete message body."""
+
+
 class ResourceExistsError(HttpResponseError):
     """An error response with status code 4xx.
     This will not be raised directly by the Azure core pipeline."""

sdk/core/azure-core/azure/core/pipeline/transport/_aiohttp.py

@@ -40,7 +40,7 @@
 from multidict import CIMultiDict
 
 from azure.core.configuration import ConnectionConfiguration
-from azure.core.exceptions import ServiceRequestError, ServiceResponseError
+from azure.core.exceptions import ServiceRequestError, ServiceResponseError, IncompleteReadError
 from azure.core.pipeline import Pipeline
 
 from ._base import HttpRequest
@@ -304,6 +304,10 @@ async def __anext__(self):
         except _ResponseStopIteration:
             internal_response.close()
             raise StopAsyncIteration()
+        except aiohttp.client_exceptions.ClientPayloadError as err:
+            _LOGGER.warning("Incomplete download: %s", err)
+            internal_response.close()
+            raise IncompleteReadError(err, error=err)
         except Exception as err:
             _LOGGER.warning("Unable to stream download: %s", err)
             internal_response.close()
@@ -381,7 +385,10 @@ def text(self, encoding: Optional[str] = None) -> str:
 
     async def load_body(self) -> None:
         """Load in memory the body, so it could be accessible from sync methods."""
-        self._content = await self.internal_response.read()
+        try:
+            self._content = await self.internal_response.read()
+        except aiohttp.client_exceptions.ClientPayloadError as err:
+            raise IncompleteReadError(err, error=err)
 
     def stream_download(self, pipeline, **kwargs) -> AsyncIteratorType[bytes]:
         """Generator for streaming response body data.

sdk/core/azure-core/azure/core/pipeline/transport/_requests_asyncio.py

@@ -36,7 +36,8 @@
 
 from azure.core.exceptions import (
     ServiceRequestError,
-    ServiceResponseError
+    ServiceResponseError,
+    IncompleteReadError,
 )
 from azure.core.pipeline import Pipeline
 from ._base import HttpRequest
@@ -134,7 +135,7 @@ async def send(self, request, **kwargs):  # pylint:disable=invalid-overridden-me
         self.open()
         loop = kwargs.get("loop", _get_running_loop())
         response = None
-        error = None # type: Optional[Union[ServiceRequestError, ServiceResponseError]]
+        error = None # type: Optional[Union[ServiceRequestError, ServiceResponseError, IncompleteReadError]]
         data_to_send = await self._retrieve_request_data(request)
         try:
             response = await loop.run_in_executor(
@@ -151,6 +152,7 @@ async def send(self, request, **kwargs):  # pylint:disable=invalid-overridden-me
                     cert=kwargs.pop('connection_cert', self.connection_config.cert),
                     allow_redirects=False,
                     **kwargs))
+            response.raw.enforce_content_length = True
 
         except urllib3.exceptions.NewConnectionError as err:
             error = ServiceRequestError(err, error=err)
@@ -161,6 +163,8 @@ async def send(self, request, **kwargs):  # pylint:disable=invalid-overridden-me
                 error = ServiceResponseError(err, error=err)
             else:
                 error = ServiceRequestError(err, error=err)
+        except requests.exceptions.ChunkedEncodingError as err:
+            error = IncompleteReadError(err, error=err)
         except requests.RequestException as err:
             error = ServiceRequestError(err, error=err)
 
@@ -223,6 +227,10 @@ async def __anext__(self):
             raise StopAsyncIteration()
         except requests.exceptions.StreamConsumedError:
             raise
+        except requests.exceptions.ChunkedEncodingError as err:
+            _LOGGER.warning("Incomplete download: %s", err)
+            internal_response.close()
+            raise IncompleteReadError(err, error=err)
         except Exception as err:
             _LOGGER.warning("Unable to stream download: %s", err)
             internal_response.close()

sdk/core/azure-core/azure/core/pipeline/transport/_requests_basic.py

@@ -36,7 +36,8 @@
 from azure.core.configuration import ConnectionConfiguration
 from azure.core.exceptions import (
     ServiceRequestError,
-    ServiceResponseError
+    ServiceResponseError,
+    IncompleteReadError,
 )
 from . import HttpRequest # pylint: disable=unused-import
 
@@ -79,6 +80,7 @@ def _read_raw_stream(response, chunk_size=1):
     # https://github.com/psf/requests/blob/master/requests/models.py#L774
     response._content_consumed = True  # pylint: disable=protected-access
 
+
 class _RequestsTransportResponseBase(_HttpResponseBase):
     """Base class for accessing response data.
 
@@ -164,6 +166,10 @@ def __next__(self):
             raise StopIteration()
         except requests.exceptions.StreamConsumedError:
             raise
+        except requests.exceptions.ChunkedEncodingError as err:
+            _LOGGER.warning("Incomplete download: %s", err)
+            internal_response.close()
+            raise IncompleteReadError(err, error=err)
         except Exception as err:
             _LOGGER.warning("Unable to stream download: %s", err)
             internal_response.close()
@@ -289,7 +295,7 @@ def send(self, request, **kwargs): # type: ignore
         """
         self.open()
         response = None
-        error = None # type: Optional[Union[ServiceRequestError, ServiceResponseError]]
+        error = None # type: Optional[Union[ServiceRequestError, ServiceResponseError, IncompleteReadError]]
 
         try:
             connection_timeout = kwargs.pop('connection_timeout', self.connection_config.timeout)
@@ -313,6 +319,7 @@ def send(self, request, **kwargs): # type: ignore
                 cert=kwargs.pop('connection_cert', self.connection_config.cert),
                 allow_redirects=False,
                 **kwargs)
+            response.raw.enforce_content_length = True
 
         except (urllib3.exceptions.NewConnectionError, urllib3.exceptions.ConnectTimeoutError) as err:
             error = ServiceRequestError(err, error=err)
@@ -323,6 +330,8 @@ def send(self, request, **kwargs): # type: ignore
                 error = ServiceResponseError(err, error=err)
             else:
                 error = ServiceRequestError(err, error=err)
+        except requests.exceptions.ChunkedEncodingError as err:
+            error = IncompleteReadError(err, error=err)
         except requests.RequestException as err:
             error = ServiceRequestError(err, error=err)
 

sdk/core/azure-core/azure/core/pipeline/transport/_requests_trio.py

@@ -36,7 +36,8 @@
 
 from azure.core.exceptions import (
     ServiceRequestError,
-    ServiceResponseError
+    ServiceResponseError,
+    IncompleteReadError,
 )
 from azure.core.pipeline import Pipeline
 from ._base import HttpRequest
@@ -105,6 +106,10 @@ async def __anext__(self):
             raise StopAsyncIteration()
         except requests.exceptions.StreamConsumedError:
             raise
+        except requests.exceptions.ChunkedEncodingError as err:
+            _LOGGER.warning("Incomplete download: %s", err)
+            internal_response.close()
+            raise IncompleteReadError(err, error=err)
         except Exception as err:
             _LOGGER.warning("Unable to stream download: %s", err)
             internal_response.close()
@@ -184,7 +189,7 @@ async def send(self, request, **kwargs: Any):  # pylint:disable=invalid-overridd
         self.open()
         trio_limiter = kwargs.get("trio_limiter", None)
         response = None
-        error = None # type: Optional[Union[ServiceRequestError, ServiceResponseError]]
+        error = None # type: Optional[Union[ServiceRequestError, ServiceResponseError, IncompleteReadError]]
         data_to_send = await self._retrieve_request_data(request)
         try:
             try:
@@ -217,6 +222,7 @@ async def send(self, request, **kwargs: Any):  # pylint:disable=invalid-overridd
                         allow_redirects=False,
                         **kwargs),
                     limiter=trio_limiter)
+            response.raw.enforce_content_length = True
 
         except urllib3.exceptions.NewConnectionError as err:
             error = ServiceRequestError(err, error=err)
@@ -227,6 +233,8 @@ async def send(self, request, **kwargs: Any):  # pylint:disable=invalid-overridd
                 error = ServiceResponseError(err, error=err)
             else:
                 error = ServiceRequestError(err, error=err)
+        except requests.exceptions.ChunkedEncodingError as err:
+            error = IncompleteReadError(err, error=err)
         except requests.RequestException as err:
             error = ServiceRequestError(err, error=err)
 

sdk/core/azure-core/tests/async_tests/test_content_length_checking_async.py

@@ -0,0 +1,28 @@
+# coding: utf-8
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See LICENSE.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+from azure.core.pipeline import AsyncPipeline
+from azure.core.pipeline.transport import (
+    HttpRequest,
+)
+from azure.core import AsyncPipelineClient
+from azure.core.exceptions import IncompleteReadError
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_aio_transport_short_read_download_stream(port):
+    url = "http://localhost:{}/errors/short-data".format(port)
+    client = AsyncPipelineClient(url)
+    with pytest.raises(IncompleteReadError):
+        async with client:
+            request = HttpRequest("GET", url)
+            pipeline_response = await client._pipeline.run(request, stream=True)
+            response = pipeline_response.http_response
+            data = response.stream_download(client._pipeline)
+            content = b""
+            async for d in data:
+                content += d

sdk/core/azure-core/tests/test_content_length_checking.py

@@ -0,0 +1,27 @@
+# coding: utf-8
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See LICENSE.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+from azure.core import PipelineClient
+from azure.core.pipeline import Pipeline
+from azure.core.pipeline.transport import (
+    HttpRequest,
+    RequestsTransport,
+)
+from azure.core.exceptions import IncompleteReadError
+import pytest
+
+
+def test_sync_transport_short_read_download_stream(port):
+    url = "http://localhost:{}/errors/short-data".format(port)
+    client = PipelineClient(url)
+    request = HttpRequest("GET", url)
+    with pytest.raises(IncompleteReadError):
+        pipeline_response = client._pipeline.run(request, stream=True)
+        response = pipeline_response.http_response
+        data = response.stream_download(client._pipeline)
+        content = b""
+        for d in data:
+            content += d

sdk/core/azure-core/tests/testserver_tests/coretestserver/coretestserver/test_routes/errors.py

@@ -26,3 +26,10 @@ def __iter__(self):
             yield b"Hello, "
             yield b"world!"
     return Response(StreamingBody(), status=500)
+
+@errors_api.route('/short-data', methods=['GET'])
+def get_short_data():
+    response = Response(b"X" * 4, status=200)
+    response.automatically_set_content_length = False
+    response.headers["Content-Length"] = "8"
+    return response