CVE-2015-5263: Improper Certificate Validation (CWE-295)
value: pulp-consumer-client 2.4.0 through 2.6.3 does not check the server's TLS certificate signatures when retrieving the server's public key upon registration.
CWE description: The product does not validate, or incorrectly validates, a certificate.
CVSS v3 base score: 8.1
Published: 2017-09-25T21:29Z
Repository: https://github.com/pulp/pulp
Language: Python
Files changed: bindings/pulp/bindings/bindings.py, bindings/pulp/bindings/consumer.py, bindings/pulp/bindings/server.py, bindings/pulp/bindings/static.py, bindings/test/unit/test_static.py, client_consumer/pulp/client/consumer/cli.py, client_consumer/test/unit/test_cli.py
Commit date: 2015-09-16 17:43:29 -0400
Commit message: convert consumer register to bindings

Diff: diff --git a/bindings/pulp/bindings/bindings.py b/bindings/pulp/bindings/bindings.py
index 4e4dbf95d..9d5b76ad0 100644
--- a/bindings/pulp/bindings/bindings.py
+++ b/bindings/pulp/bindings/bindings.py
@@ -17,10 +17,11 @@ from pulp.bindings.event_listeners import EventListenerAPI
 from pulp.bindings.repo_groups import *
 from pulp.bindings.repository import *
 from pulp.bindings.consumer_groups import *
 from pulp.bindings.consumer import *
 from pulp.bindings.server_info import ServerInfoAPI
+from pulp.bindings.static import StaticRequest
 from pulp.bindings.tasks import TasksAPI, TaskSearchAPI
 from pulp.bindings.upload import UploadAPI
 from pulp.bindings.auth import *
 
 
@@ -65,10 +66,11 @@ class Bindings(object):
         self.repo_search = RepositorySearchAPI(pulp_connection)
         self.repo_sync_schedules = RepositorySyncSchedulesAPI(pulp_connection)
         self.repo_unit = RepositoryUnitAPI(pulp_connection)
         self.role = RoleAPI(pulp_connection)
         self.server_info = ServerInfoAPI(pulp_connection)
+        self.static = StaticRequest(pulp_connection)
         self.tasks = TasksAPI(pulp_connection)
         self.tasks_search = TaskSearchAPI(pulp_connection)
         self.uploads = UploadAPI(pulp_connection)
         self.user = UserAPI(pulp_connection)
         self.user_search = UserSearchAPI(pulp_connection)
diff --git a/bindings/pulp/bindings/consumer.py b/bindings/pulp/bindings/consumer.py
index 0c07d448c..f59f8361f 100644
--- a/bindings/pulp/bindings/consumer.py
+++ b/bindings/pulp/bindings/consumer.py
@@ -109,11 +109,11 @@ class ConsumerContentSchedulesAPI(PulpAPI):
         return self.server.GET(url)
 
     def get_schedule(self, action, consumer_id, schedule_id):
         url = self.base_path % consumer_id + action + '/%s/' % schedule_id
         return self.server.GET(url)
-    
+
     def add_schedule(self, action, consumer_id, schedule, units, failure_threshold=UNSPECIFIED,
                      enabled=UNSPECIFIED, options=UNSPECIFIED):
         url = self.base_path % consumer_id + action + '/'
         body = {
             'schedule' : schedule,
@@ -123,11 +123,11 @@ class ConsumerContentSchedulesAPI(PulpAPI):
             'options': options,
             }
         # Strip out anything that wasn't specified by the caller
         body = dict([(k, v) for k, v in body.items() if v is not UNSPECIFIED])
         return self.server.POST(url, body)
- 
+
     def delete_schedule(self, action, consumer_id, schedule_id):
         url = self.base_path % consumer_id + action + '/%s/' % schedule_id
         return self.server.DELETE(url)
 
     def update_schedule(self, action, consumer_id, schedule_id, schedule=UNSPECIFIED, units=UNSPECIFIED,
@@ -154,21 +154,21 @@ class BindingsAPI(PulpAPI):
     def find_by_id(self, consumer_id, repo_id=None):
         path = self.BASE_PATH % consumer_id
         if repo_id:
             path += '%s/' % repo_id
         return self.server.GET(path)
-    
+
     def bind(self, consumer_id, repo_id, distributor_id, notify_agent=True, binding_config=None):
         path = self.BASE_PATH % consumer_id
         data = {
             'repo_id' :repo_id,
             'distributor_id' :distributor_id,
             'notify_agent': notify_agent,
             'binding_config': binding_config or {}
         }
         return self.server.POST(path, data)
-    
+
     def unbind(self, consumer_id, repo_id, distributor_id, force=False):
         path = self.BASE_PATH % consumer_id + "%s/" % repo_id + "%s/" % distributor_id
         body = dict(force=force)
         return self.server.DELETE(path, body)
 
diff --git a/bindings/pulp/bindings/server.py b/bindings/pulp/bindings/server.py
index f0782d018..9a46795fd 100644
--- a/bindings/pulp/bindings/server.py
+++ b/bindings/pulp/bindings/server.py
@@ -82,31 +82,33 @@ class PulpConnection(object):
 
         # SSL validation settings
         self.verify_ssl = verify_ssl
         self.ca_path = ca_path
 
-    def DELETE(self, path, body=None, log_request_body=True):
-        return self._request('DELETE', path, body=body, log_request_body=log_request_body)
+    def DELETE(self, path, body=None, log_request_body=True, ignore_prefix=False):
+        return self._request('DELETE', path, body=body, log_request_body=log_request_body,
+                             ignore_prefix=ignore_prefix)
 
-    def GET(self, path, queries=()):
-        return self._request('GET', path, queries)
+    def GET(self, path, queries=(), ignore_prefix=False):
+        return self._request('GET', path, queries, ignore_prefix=ignore_prefix)
 
-    def HEAD(self, path):
-        return self._request('HEAD', path)
+    def HEAD(self, path, ignore_prefix=False):
+        return self._request('HEAD', path, ignore_prefix=ignore_prefix)
 
-    def POST(self, path, body=None, ensure_encoding=True, log_request_body=True):
+    def POST(self, path, body=None, ensure_encoding=True, log_request_body=True,
+             ignore_prefix=False):
         return self._request('POST', path, body=body, ensure_encoding=ensure_encoding,
-                             log_request_body=log_request_body)
+                             log_request_body=log_request_body, ignore_prefix=ignore_prefix)
 
-    def PUT(self, path, body, ensure_encoding=True, log_request_body=True):
+    def PUT(self, path, body, ensure_encoding=True, log_request_body=True, ignore_prefix=False):
         return self._request('PUT', path, body=body, ensure_encoding=ensure_encoding,
-                             log_request_body=log_request_body)
+                             log_request_body=log_request_body, ignore_prefix=ignore_prefix)
 
     # protected request utilities ---------------------------------------------
 
     def _request(self, method, path, queries=(), body=None, ensure_encoding=True,
-                 log_request_body=True):
+                 log_request_body=True, ignore_prefix=False):
         """
         make a HTTP request to the pulp server and return the response
 
         :param method:  name of an HTTP method such as GET, POST, PUT, HEAD
                         or DELETE
@@ -128,18 +130,21 @@ class PulpConnection(object):
         :type ensure_encoding: bool
 
         :param log_request_body: Toggle logging of the request body, defaults to true
         :type log_request_body: bool
 
+        :param ignore_prefix: when building the url, disregard the self.path_prefix
+        :type  ignore_prefix: bool
+
         :return:    Response object
         :rtype:     pulp.bindings.responses.Response
 
         :raises:    ConnectionException or one of the RequestExceptions
                     (depending on response codes) in case of unsuccessful
                     request
         """
-        url = self._build_url(path, queries)
+        url = self._build_url(path, queries, ignore_prefix)
         if ensure_encoding:
             body = self._process_body(body)
         if not isinstance(body, (NoneType, basestring)):
             body = json.dumps(body)
         self.log.debug('sending %s request to %s' % (method, url))
@@ -199,11 +204,11 @@ class PulpConnection(object):
                 raise exceptions.PulpServerException(response_body)
 
         else:
             raise code_class_mappings[response_code](response_body)
 
-    def _build_url(self, path, queries=()):
+    def _build_url(self, path, queries, ignore_prefix):
         """
         Takes a relative path and query parameters, combines them with the
         base path, and returns the result. Handles utf-8 encoding as necessary.
 
         :param path:    relative path for this request, relative to
@@ -215,17 +220,19 @@ class PulpConnection(object):
 
         :param queries: mapping object or a sequence of 2-element tuples,
                         in either case representing key-value pairs to be used
                         as query parameters on the URL.
         :type  queries: mapping object or sequence of 2-element tuples
+        :param ignore_prefix: when building the url, disregard the self.path_prefix
+        :type  ignore_prefix: bool
 
         :return:    path that is a composite of self.path_prefix, path, and
                     queries. May be relative or absolute depending on the nature
                     of self.path_prefix
         """
         # build the request url from the path and queries dict or tuple
-        if not path.startswith(self.path_prefix):
+        if not path.startswith(self.path_prefix) and not ignore_prefix:
             if path.startswith('/'):
                 path = path[1:]
             path = '/'.join((self.path_prefix, path))
             # Check if path is ascii and uses appropriate characters,
             # else convert to binary or unicode as necessary.
diff --git a/bindings/pulp/bindings/static.py b/bindings/pulp/bindings/static.py
new file mode 100644
index 000000000..3e2e2679e
--- /dev/null
+++ b/bindings/pulp/bindings/static.py
@@ -0,0 +1,16 @@
+from pulp.bindings.base import PulpAPI
+
+
+class StaticRequest(PulpAPI):
+    """
+    Connection class to access static calls
+    """
+
+    def get_server_key(self):
+        """
+        Retrieve the server's public key.
+
+        :return: rsa public key
+        :rtype:  str
+        """
+        return self.server.GET('/pulp/static/rsa_pub.key', ignore_prefix=True)
diff --git a/bindings/test/unit/test_static.py b/bindings/test/unit/test_static.py
new file mode 100644
index 000000000..1d991f453
--- /dev/null
+++ b/bindings/test/unit/test_static.py
@@ -0,0 +1,22 @@
+import unittest
+
+import mock
+
+from pulp.bindings.server import PulpConnection
+from pulp.bindings.static import StaticRequest
+
+
+class TestStaticRequest(unittest.TestCase):
+    """
+    Tests for static file requests.
+    """
+
+    def test_get_server_key(self):
+        """
+        Test that the correct path is given to the binding.
+        """
+        static_request = StaticRequest(mock.MagicMock(spec=PulpConnection))
+        response = static_request.get_server_key()
+        static_request.server.GET.assert_called_once_with('/pulp/static/rsa_pub.key',
+                                                          ignore_prefix=True)
+        self.assertTrue(response is static_request.server.GET.return_value)
diff --git a/client_consumer/pulp/client/consumer/cli.py b/client_consumer/pulp/client/consumer/cli.py
index ea9ab22dd..d387ba0ae 100644
--- a/client_consumer/pulp/client/consumer/cli.py
+++ b/client_consumer/pulp/client/consumer/cli.py
@@ -89,47 +89,46 @@ def initialize(context):
 
     d = 'displays the registration status of this consumer'
     context.cli.add_command(StatusCommand(context, 'status', _(d)))
 
 
-def download(url, location):
+def write_to_location(location, content):
     """
-    Download files to the specified location.
-    :param url: The file URL.
-    :type url: str
-    :param location: The absolute path to where the downloaded
-        file is to be stored.
-    :type location: str
+    Write content to a path. Ensures that the entire path exists, creating directories if necessary.
+
+    :param location: path that should exist
+    :type  location: str
+    :param content: bits to be written to file
+    :type  content: str
     """
-    request = urllib2.urlopen(url)
     try:
-        content = request.read()
+        os.makedirs(os.path.dirname(location))
+    except OSError, e:
+        if e.errno != errno.EEXIST:
+            raise
+    try:
         fp = open(location, 'w+')
-        try:
-            fp.write(content)
-        finally:
-            fp.close()
+        fp.write(content)
     finally:
-        request.close()
+        fp.close()
 
 
-def update_server_key(conf):
+def update_server_key(command_inst):
     """
-    Download the server's RSA key and store in the location
-    specified in the configuration.
-    :param conf: The consumer configuration object.
-    :type conf: dict
+    Ensure that the server's public key stored on the consumer is up to date.
+
+    :param command_inst: instance of a CLI command
+    :type  command_inst: pulp.client.extensions.extensions.PulpCliCommand
     """
-    host = conf['server']['host']
-    location = conf['server']['rsa_pub']
-    url = 'https://%s/pulp/static/rsa_pub.key' % host
     try:
-        os.makedirs(os.path.dirname(location))
-    except OSError, e:
-        if e.errno != errno.EEXIST:
-            raise
-    download(url, location)
+        key_reply = command_inst.context.server.static.get_server_key()
+    except Exception, e:
+        msg = _('Download server RSA key failed [%(e)s]' % {'e': e})
+        command_inst.prompt.render_failure_message(msg)
+    else:
+        key_location = command_inst.context.config['server']['rsa_pub']
+        write_to_location(key_location, key_reply.response_body)
 
 
 # -- common exceptions --------------------------------------------------------
 
 
@@ -199,18 +198,11 @@ class RegisterCommand(PulpCliCommand):
         try:
             fp.write(certificate)
         finally:
             fp.close()
 
-        # download server public key
-
-        try:
-            update_server_key(self.context.config)
-        except Exception, e:
-            msg = _('Download server RSA key failed [%(e)s]' % {'e': e})
-            self.prompt.render_failure_message(msg)
-
+        update_server_key(self)
         self.prompt.render_success_message('Consumer [%s] successfully registered' % consumer_id)
 
 
 class UpdateCommand(PulpCliCommand):
 
@@ -248,15 +240,11 @@ class UpdateCommand(PulpCliCommand):
         try:
             self.context.server.consumer.update(consumer_id, delta)
             self.prompt.render_success_message('Consumer [%s] successfully updated' % consumer_id)
             if not kwargs.get(OPTION_EXCHANGE_KEYS.keyword):
                 return
-            try:
-                update_server_key(self.context.config)
-            except Exception, e:
-                msg = _('Download server RSA key failed [%(e)s]' % {'e': e})
-                self.prompt.render_failure_message(msg)
+            update_server_key(self)
         except NotFoundException:
             self.prompt.write('Consumer [%s] does not exist on the server' % consumer_id, tag='not-found')
 
 
 class UnregisterCommand(PulpCliCommand):
diff --git a/client_consumer/test/unit/test_cli.py b/client_consumer/test/unit/test_cli.py
new file mode 100644
index 000000000..ff0b0eec3
--- /dev/null
+++ b/client_consumer/test/unit/test_cli.py
@@ -0,0 +1,116 @@
+import unittest
+
+import mock
+
+from pulp.client.consumer import cli
+
+
+class TestWriteToLocation(unittest.TestCase):
+    """
+    Tests for writing content to a file.
+    """
+
+    @mock.patch('pulp.client.consumer.cli.os')
+    @mock.patch('__builtin__.open', new_callable=mock.MagicMock())
+    def test_dir_struct_exists(self, mopen, mock_os):
+        """
+        Test that when the directory structure already exists, the write still happens.
+        """
+
+        class MockException(OSError):
+            pass
+
+        mock_e = MockException()
+        mock_e.errno = 17
+        mock_os.makedirs.side_effect = mock_e
+        mock_fp = open.return_value
+
+        cli.write_to_location('test/loc', 'content')
+        mock_os.path.dirname.assert_called_once_with('test/loc')
+        mock_os.makedirs.assert_called_once_with(mock_os.path.dirname.return_value)
+        open.assert_called_once_with('test/loc', 'w+')
+        mock_fp.write.assert_called_once_with('content')
+        mock_fp.close.assert_called_once_with()
+
+    @mock.patch('pulp.client.consumer.cli.os')
+    @mock.patch('__builtin__.open', new_callable=mock.MagicMock())
+    def test_misc_os_err(self, mopen, mock_os):
+        """
+        Test that misc errors are reraised and the write does not happen.
+        """
+
+        class MockException(OSError):
+            pass
+
+        mock_e = MockException()
+        mock_e.errno = 16
+        mock_os.makedirs.side_effect = mock_e
+        mock_fp = open.return_value
+
+        self.assertRaises(MockException, cli.write_to_location, 'test/loc', 'content')
+        self.assertEqual(mock_fp.write.call_count, 0)
+
+    @mock.patch('pulp.client.consumer.cli.os')
+    @mock.patch('__builtin__.open', new_callable=mock.MagicMock())
+    def test_write_err(self, *_):
+        """
+        If there is a problem with the write, the file is still closed.
+        """
+
+        class MockException(Exception):
+            pass
+
+        mock_fp = open.return_value
+        mock_fp.write.side_effect = MockException
+        self.assertRaises(MockException, cli.write_to_location, 'test/loc', 'content')
+        mock_fp.write.assert_called_once_with('content')
+        mock_fp.close.assert_called_once_with()
+
+    @mock.patch('pulp.client.consumer.cli.os')
+    @mock.patch('__builtin__.open', new_callable=mock.MagicMock())
+    def test_as_expected(self, mopen, mock_os):
+        """
+        When everything works as expected, ensure that the file is closed.
+        """
+        mock_fp = open.return_value
+        cli.write_to_location('test/loc', 'content')
+        mock_os.path.dirname.assert_called_once_with('test/loc')
+        mock_os.makedirs.assert_called_once_with(mock_os.path.dirname.return_value)
+        open.assert_called_once_with('test/loc', 'w+')
+        mock_fp.write.assert_called_once_with('content')
+        mock_fp.close.assert_called_once_with()
+
+
+class TestUpdateServerKey(unittest.TestCase):
+    """
+    Tests for updating the server key.
+    """
+
+    @mock.patch('pulp.client.consumer.cli.write_to_location')
+    def test_as_expected(self, mock_write):
+        """
+        Everything is as expected, content is written to the location in the config file.
+        """
+        mock_cmd = mock.MagicMock()
+        key_response = mock_cmd.context.server.static.get_server_key.return_value
+        key_loc = mock_cmd.context.config['getter']['getter']
+        cli.update_server_key(mock_cmd)
+        mock_write.assert_called_once_with(key_loc, key_response.response_body)
+
+    @mock.patch('pulp.client.consumer.cli.write_to_location')
+    def test_binding_exception(self, mock_write):
+        """
+        If there is a problem getting the key, do not attempt to write the file.
+        """
+
+        class MockException(Exception):
+
+            def __str__(self):
+                return "Mock Exception str"
+
+        mock_cmd = mock.MagicMock()
+        mock_cmd.context.server.static.get_server_key.side_effect = MockException()
+        cli.update_server_key(mock_cmd)
+        msg = 'Download server RSA key failed [Mock Exception str]'
+        mock_cmd.prompt.render_failure_message.assert_called_once_with(msg)
+        self.assertEqual(mock_write.call_count, 0)

Vulnerable code:     
 
    
    
    def DELETE(self, path, body=None, log_request_body=True):
        return self._request('DELETE', path, body=body, log_request_body=log_request_body)
    def GET(self, path, queries=()):
        return self._request('GET', path, queries)
    def HEAD(self, path):
        return self._request('HEAD', path)
    def POST(self, path, body=None, ensure_encoding=True, log_request_body=True):
                             log_request_body=log_request_body)
    def PUT(self, path, body, ensure_encoding=True, log_request_body=True):
                             log_request_body=log_request_body)
                 log_request_body=True):
        url = self._build_url(path, queries)
    def _build_url(self, path, queries=()):
        if not path.startswith(self.path_prefix):
def download(url, location):
    Download files to the specified location.
    :param url: The file URL.
    :type url: str
    :param location: The absolute path to where the downloaded
        file is to be stored.
    :type location: str
    request = urllib2.urlopen(url)
        content = request.read()
        try:
            fp.write(content)
        finally:
            fp.close()
        request.close()
def update_server_key(conf):
    Download the server's RSA key and store in the location
    specified in the configuration.
    :param conf: The consumer configuration object.
    :type conf: dict
    host = conf['server']['host']
    location = conf['server']['rsa_pub']
    url = 'https://%s/pulp/static/rsa_pub.key' % host
        os.makedirs(os.path.dirname(location))
    except OSError, e:
        if e.errno != errno.EEXIST:
            raise
    download(url, location)
        # download server public key

        try:
            update_server_key(self.context.config)
        except Exception, e:
            msg = _('Download server RSA key failed [%(e)s]' % {'e': e})
            self.prompt.render_failure_message(msg)

            try:
                update_server_key(self.context.config)
            except Exception, e:
                msg = _('Download server RSA key failed [%(e)s]' % {'e': e})
                self.prompt.render_failure_message(msg)
Fixed code: from pulp.bindings.static import StaticRequest
        self.static = StaticRequest(pulp_connection)




    def DELETE(self, path, body=None, log_request_body=True, ignore_prefix=False):
        return self._request('DELETE', path, body=body, log_request_body=log_request_body,
                             ignore_prefix=ignore_prefix)
    def GET(self, path, queries=(), ignore_prefix=False):
        return self._request('GET', path, queries, ignore_prefix=ignore_prefix)
    def HEAD(self, path, ignore_prefix=False):
        return self._request('HEAD', path, ignore_prefix=ignore_prefix)
    def POST(self, path, body=None, ensure_encoding=True, log_request_body=True,
             ignore_prefix=False):
                             log_request_body=log_request_body, ignore_prefix=ignore_prefix)
    def PUT(self, path, body, ensure_encoding=True, log_request_body=True, ignore_prefix=False):
                             log_request_body=log_request_body, ignore_prefix=ignore_prefix)
                 log_request_body=True, ignore_prefix=False):
        :param ignore_prefix: when building the url, disregard the self.path_prefix
        :type  ignore_prefix: bool

        url = self._build_url(path, queries, ignore_prefix)
    def _build_url(self, path, queries, ignore_prefix):
        :param ignore_prefix: when building the url, disregard the self.path_prefix
        :type  ignore_prefix: bool
        if not path.startswith(self.path_prefix) and not ignore_prefix:
from pulp.bindings.base import PulpAPI


class StaticRequest(PulpAPI):
    """
    Connection class to access static calls
    """

    def get_server_key(self):
        """
        Retrieve the server's public key.

        :return: rsa public key
        :rtype:  str
        """
        return self.server.GET('/pulp/static/rsa_pub.key', ignore_prefix=True)
import unittest

import mock

from pulp.bindings.server import PulpConnection
from pulp.bindings.static import StaticRequest


class TestStaticRequest(unittest.TestCase):
    """
    Tests for static file requests.
    """

    def test_get_server_key(self):
        """
        Test that the correct path is given to the binding.
        """
        static_request = StaticRequest(mock.MagicMock(spec=PulpConnection))
        response = static_request.get_server_key()
        static_request.server.GET.assert_called_once_with('/pulp/static/rsa_pub.key',
                                                          ignore_prefix=True)
        self.assertTrue(response is static_request.server.GET.return_value)
def write_to_location(location, content):
    Write content to a path. Ensures that the entire path exists, creating directories if necessary.

    :param location: path that should exist
    :type  location: str
    :param content: bits to be written to file
    :type  content: str
        os.makedirs(os.path.dirname(location))
    except OSError, e:
        if e.errno != errno.EEXIST:
            raise
    try:
        fp.write(content)
        fp.close()
def update_server_key(command_inst):
    Ensure that the server's public key stored on the consumer is up to date.

    :param command_inst: instance of a CLI command
    :type  command_inst: pulp.client.extensions.extensions.PulpCliCommand
        key_reply = command_inst.context.server.static.get_server_key()
    except Exception, e:
        msg = _('Download server RSA key failed [%(e)s]' % {'e': e})
        command_inst.prompt.render_failure_message(msg)
    else:
        key_location = command_inst.context.config['server']['rsa_pub']
        write_to_location(key_location, key_reply.response_body)
        update_server_key(self)
            update_server_key(self)
import unittest

import mock

from pulp.client.consumer import cli


class TestWriteToLocation(unittest.TestCase):
    """
    Tests for writing content to a file.
    """

    @mock.patch('pulp.client.consumer.cli.os')
    @mock.patch('__builtin__.open', new_callable=mock.MagicMock())
    def test_dir_struct_exists(self, mopen, mock_os):
        """
        Test that when the directory structure already exists, the write still happens.
        """

        class MockException(OSError):
            pass

        mock_e = MockException()
        mock_e.errno = 17
        mock_os.makedirs.side_effect = mock_e
        mock_fp = open.return_value

        cli.write_to_location('test/loc', 'content')
        mock_os.path.dirname.assert_called_once_with('test/loc')
        mock_os.makedirs.assert_called_once_with(mock_os.path.dirname.return_value)
        open.assert_called_once_with('test/loc', 'w+')
        mock_fp.write.assert_called_once_with('content')
        mock_fp.close.assert_called_once_with()

    @mock.patch('pulp.client.consumer.cli.os')
    @mock.patch('__builtin__.open', new_callable=mock.MagicMock())
    def test_misc_os_err(self, mopen, mock_os):
        """
        Test that misc errors are reraised and the write does not happen.
        """

        class MockException(OSError):
            pass

        mock_e = MockException()
        mock_e.errno = 16
        mock_os.makedirs.side_effect = mock_e
        mock_fp = open.return_value

        self.assertRaises(MockException, cli.write_to_location, 'test/loc', 'content')
        self.assertEqual(mock_fp.write.call_count, 0)

    @mock.patch('pulp.client.consumer.cli.os')
    @mock.patch('__builtin__.open', new_callable=mock.MagicMock())
    def test_write_err(self, *_):
        """
        If there is a problem with the write, the file is still closed.
        """

        class MockException(Exception):
            pass

        mock_fp = open.return_value
        mock_fp.write.side_effect = MockException
        self.assertRaises(MockException, cli.write_to_location, 'test/loc', 'content')
        mock_fp.write.assert_called_once_with('content')
        mock_fp.close.assert_called_once_with()

    @mock.patch('pulp.client.consumer.cli.os')
    @mock.patch('__builtin__.open', new_callable=mock.MagicMock())
    def test_as_expected(self, mopen, mock_os):
        """
        When everything works as expected, ensure that the file is closed.
        """
        mock_fp = open.return_value
        cli.write_to_location('test/loc', 'content')
        mock_os.path.dirname.assert_called_once_with('test/loc')
        mock_os.makedirs.assert_called_once_with(mock_os.path.dirname.return_value)
        open.assert_called_once_with('test/loc', 'w+')
        mock_fp.write.assert_called_once_with('content')
        mock_fp.close.assert_called_once_with()


class TestUpdateServerKey(unittest.TestCase):
    """
    Tests for updating the server key.
    """

    @mock.patch('pulp.client.consumer.cli.write_to_location')
    def test_as_expected(self, mock_write):
        """
        Everything is as expected, content is written to the location in the config file.
        """
        mock_cmd = mock.MagicMock()
        key_response = mock_cmd.context.server.static.get_server_key.return_value
        key_loc = mock_cmd.context.config['getter']['getter']
        cli.update_server_key(mock_cmd)
        mock_write.assert_called_once_with(key_loc, key_response.response_body)

    @mock.patch('pulp.client.consumer.cli.write_to_location')
    def test_binding_exception(self, mock_write):
        """
        If there is a problem getting the key, do not attempt to write the file.
        """

        class MockException(Exception):

            def __str__(self):
                return "Mock Exception str"

        mock_cmd = mock.MagicMock()
        mock_cmd.context.server.static.get_server_key.side_effect = MockException()
        cli.update_server_key(mock_cmd)
        msg = 'Download server RSA key failed [Mock Exception str]'
        mock_cmd.prompt.render_failure_message.assert_called_once_with(msg)
        self.assertEqual(mock_write.call_count, 0)