CVE-2015-5263 (Python, published 2017-09-25T21:29Z)
Description: 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.
Severity: MEDIUM (CVSS v3 base score: 8.1)
Weakness: CWE-295 \u2014 Improper Certificate Validation
The product does not validate, or incorrectly validates, a certificate.
Repository: https://github.com/pulp/pulp
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 message: convert consumer register to bindings


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)