Check presence of errors in server response to image push

docker/api/image.py

@@ -1,8 +1,11 @@
+import itertools
+import json
 import logging
 import os
 
 from .. import auth, errors, utils
 from ..constants import DEFAULT_DATA_CHUNK_SIZE
+from ..utils.json_stream import json_stream
 
 log = logging.getLogger(__name__)
 
@@ -433,6 +436,29 @@ def pull(self, repository, tag=None, stream=False, auth_config=None,
 
         return self._result(response)
 
+    @staticmethod
+    def _raise_if_error(chunk, response):
+        """
+        Raise an exception if the given chunk of the JSON server response is a
+        dictionary and contains an "error" field. Otherwise, return the chunk
+        as-is.
+
+        Args:
+            chunk (object): A chunk of the server response.
+            response (Response): The full server response. This will be attached
+                to the exception in the event that chunk indicates an error.
+
+        Returns:
+            (object): The input chunk.
+
+        Raises:
+            :py:class:`docker.errors.APIError`
+                If the chunk of the server response contains an error message.
+        """
+        if isinstance(chunk, dict) and 'error' in chunk:
+            raise errors.APIError(chunk['error'], response=response)
+        return chunk
+
     def push(self, repository, tag=None, stream=False, auth_config=None,
              decode=False):
         """
@@ -494,8 +520,25 @@ def push(self, repository, tag=None, stream=False, auth_config=None,
 
         self._raise_for_status(response)
 
+        # The server response might have status code 200 (OK) even though the
+        # push operation has failed. To detect errors, inspect each JSON chunk
+        # of the server response and check if an "error" entry is present.
+        # See: https://github.com/docker/docker-py/issues/3277
         if stream:
-            return self._stream_helper(response, decode=decode)
+            if decode:
+                return (self._raise_if_error(chunk, response) for chunk in
+                        self._stream_helper(response, decode=True))
+            else:
+                result_stream, internal_stream = itertools.tee(
+                    self._stream_helper(response, decode=False))
+                for chunk_json in json_stream(internal_stream):
+                    self._raise_if_error(chunk_json, response)
+                return result_stream
+
+        for chunk_str in response.text.splitlines():
+            chunk_json = json.loads(chunk_str)
+            if 'error' in chunk_json:
+                raise errors.APIError(chunk_json['error'], response=response)
 
         return self._result(response)
 

docker/utils/json_stream.py

@@ -1,4 +1,3 @@
-import json
 import json.decoder
 
 from ..errors import StreamParseError
@@ -37,30 +36,12 @@ def json_stream(stream):
     This handles streams which are inconsistently buffered (some entries may
     be newline delimited, and others are not).
     """
-    return split_buffer(stream, json_splitter, json_decoder.decode)
-
-
-def line_splitter(buffer, separator='\n'):
-    index = buffer.find(str(separator))
-    if index == -1:
-        return None
-    return buffer[:index + 1], buffer[index + 1:]
-
-
-def split_buffer(stream, splitter=None, decoder=lambda a: a):
-    """Given a generator which yields strings and a splitter function,
-    joins all input, splits on the separator and yields each chunk.
-    Unlike string.split(), each chunk includes the trailing
-    separator, except for the last one if none was found on the end
-    of the input.
-    """
-    splitter = splitter or line_splitter
     buffered = ''
 
     for data in stream_as_text(stream):
         buffered += data
         while True:
-            buffer_split = splitter(buffered)
+            buffer_split = json_splitter(buffered)
             if buffer_split is None:
                 break
 
@@ -69,6 +50,13 @@ def split_buffer(stream, splitter=None, decoder=lambda a: a):
 
     if buffered:
         try:
-            yield decoder(buffered)
+            yield json_decoder.decode(buffered)
         except Exception as e:
             raise StreamParseError(e) from e
+
+
+def line_splitter(buffer: str, separator='\n'):
+    index = buffer.find(str(separator))
+    if index == -1:
+        return None
+    return buffer[:index + 1], buffer[index + 1:]

tests/unit/api_image_test.py

@@ -271,6 +271,33 @@ def test_push_image_with_auth(self):
             timeout=DEFAULT_TIMEOUT_SECONDS
         )
 
+
+    def test_push_image_with_auth_error(self):
+        auth_config = {
+            'username': "test_user",
+            'password': "test_password",
+            'serveraddress': "test_server",
+        }
+        encoded_auth = auth.encode_header(auth_config)
+        with pytest.raises(docker.errors.APIError, match='bad auth'):
+            self.client.push(
+                fake_api.FAKE_IMAGE_NAME_ERROR, tag=fake_api.FAKE_TAG_NAME,
+                auth_config=auth_config
+            )
+
+        fake_request.assert_called_with(
+            'POST',
+            f"{url_prefix}images/test_image_error/push",
+            params={
+                'tag': fake_api.FAKE_TAG_NAME,
+            },
+            data='{}',
+            headers={'Content-Type': 'application/json',
+                     'X-Registry-Auth': encoded_auth},
+            stream=False,
+            timeout=DEFAULT_TIMEOUT_SECONDS
+        )
+
     def test_push_image_stream(self):
         with mock.patch('docker.auth.resolve_authconfig',
                         fake_resolve_authconfig):
@@ -288,6 +315,59 @@ def test_push_image_stream(self):
             timeout=DEFAULT_TIMEOUT_SECONDS
         )
 
+
+    def test_push_image_stream_with_auth(self):
+        auth_config = {
+            'username': "test_user",
+            'password': "test_password",
+            'serveraddress': "test_server",
+        }
+        encoded_auth = auth.encode_header(auth_config)
+        self.client.push(
+            fake_api.FAKE_IMAGE_NAME, tag=fake_api.FAKE_TAG_NAME,
+            auth_config=auth_config, stream=True
+        )
+
+        fake_request.assert_called_with(
+            'POST',
+            f"{url_prefix}images/test_image/push",
+            params={
+                'tag': fake_api.FAKE_TAG_NAME,
+            },
+            data='{}',
+            headers={'Content-Type': 'application/json',
+                     'X-Registry-Auth': encoded_auth},
+            stream=True,
+            timeout=DEFAULT_TIMEOUT_SECONDS
+        )
+
+
+    def test_push_image_stream_with_auth_error(self):
+        auth_config = {
+            'username': "test_user",
+            'password': "test_password",
+            'serveraddress': "test_server",
+        }
+        encoded_auth = auth.encode_header(auth_config)
+        with pytest.raises(docker.errors.APIError, match='bad auth'):
+            self.client.push(
+                fake_api.FAKE_IMAGE_NAME_ERROR, tag=fake_api.FAKE_TAG_NAME,
+                auth_config=auth_config, stream=True
+            )
+
+        fake_request.assert_called_with(
+            'POST',
+            f"{url_prefix}images/test_image_error/push",
+            params={
+                'tag': fake_api.FAKE_TAG_NAME,
+            },
+            data='{}',
+            headers={'Content-Type': 'application/json',
+                     'X-Registry-Auth': encoded_auth},
+            stream=True,
+            timeout=DEFAULT_TIMEOUT_SECONDS
+        )
+
     def test_tag_image(self):
         self.client.tag(fake_api.FAKE_IMAGE_ID, fake_api.FAKE_REPO_NAME)
 

tests/unit/api_test.py

@@ -31,6 +31,8 @@ def response(status_code=200, content='', headers=None, reason=None, elapsed=0,
              request=None, raw=None):
     res = requests.Response()
     res.status_code = status_code
+    if isinstance(content, str):
+        content = content.encode('ascii')
     if not isinstance(content, bytes):
         content = json.dumps(content).encode('ascii')
     res._content = content
@@ -42,6 +44,30 @@ def response(status_code=200, content='', headers=None, reason=None, elapsed=0,
     return res
 
 
+def stream_response(status_code=200, content='', **kwargs):
+    if isinstance(content, str):
+        content_bytes = content.encode('ascii')
+    elif isinstance(content, bytes):
+        content_bytes = content
+    else:
+        content_bytes = json.dumps(content).encode('ascii')
+
+    body = io.BytesIO(content_bytes)
+
+    # mock a stream interface
+    raw_resp = urllib3.HTTPResponse(body=body)
+    raw_resp._fp.chunked = True
+    raw_resp._fp.chunk_left = len(body.getvalue()) - 1
+    raw_resp._fp.seek(0)
+
+    return response(
+        status_code=status_code,
+        content=content_bytes,
+        raw=raw_resp,
+        **kwargs,
+    )
+
+
 def fake_resolve_authconfig(authconfig, registry=None, *args, **kwargs):
     return None
 
@@ -59,6 +85,8 @@ def fake_resp(method, url, *args, **kwargs):
     if not key:
         raise Exception(f'{method} {url}')
     status_code, content = fake_api.fake_responses[key]()
+    if kwargs.get("stream", False):
+        return stream_response(status_code=status_code, content=content)
     return response(status_code=status_code, content=content)
 
 

tests/unit/fake_api.py

@@ -1,3 +1,5 @@
+import json
+
 from docker import constants
 
 from . import fake_stat
@@ -9,6 +11,7 @@
 FAKE_EXEC_ID = 'b098ec855f10434b5c7c973c78484208223a83f663ddaefb0f02a242840cb1c7'
 FAKE_NETWORK_ID = '1999cfb42e414483841a125ade3c276c3cb80cb3269b14e339354ac63a31b02c'
 FAKE_IMAGE_NAME = 'test_image'
+FAKE_IMAGE_NAME_ERROR = 'test_image_error'
 FAKE_TARBALL_PATH = '/path/to/tarball'
 FAKE_REPO_NAME = 'repo'
 FAKE_TAG_NAME = 'tag'
@@ -359,6 +362,12 @@ def post_fake_push():
     return status_code, response
 
 
+def post_fake_push_error():
+    status_code = 200
+    response = '{"status": "intermediate update"}\r\n{"error": "bad auth"}\r\n'
+    return status_code, response
+
+
 def post_fake_build_container():
     status_code = 200
     response = {'Id': FAKE_CONTAINER_ID}
@@ -603,6 +612,8 @@ def post_fake_config():
     get_fake_insert_image,
     f'{prefix}/{CURRENT_VERSION}/images/test_image/push':
     post_fake_push,
+    f'{prefix}/{CURRENT_VERSION}/images/test_image_error/push':
+    post_fake_push_error,
     f'{prefix}/{CURRENT_VERSION}/commit':
     post_fake_commit,
     f'{prefix}/{CURRENT_VERSION}/containers/create':