Repository: pika/pika

Issue:
Change pika to accept ssl.SSLContext instead of dict or homegrown SSLOptions attributes for 1.0.0
Pre 1.0.0, pika accepts either a dict or a homegrown `pika.connection.SSLOptions` instance and then [converts it to `ssl.SSLContext` (in case of `SSLOptions`) or passes the dict as kwargs to `ssl.wrap_socket()`](https://github.com/pika/pika/blob/09270712bf5912566df8af10ae68a1ddaea4f4e1/pika/adapters/base_connection.py#L512-L539).

I can't think of a good reason to march on to 1.0.0 with this status quo. It's unnecessary and limiting. User's should just be able to pass an `ssl.SSLContext` instance which gives them most flexibility, including the ability to cache ssl contexts. For example, `asyncio` in Python 3 accepts only `ssl.SSLContext` instances.

Proposal for pika 1.0.0:
1. Modify `pika.connection.SSLOptions` constructor to accept only a non-None `ssl.SSLContext` instance and an optional `server_hostname` arg
2. `server_hostname` defaults to None and is used for CN validation - see discussion in issue #464.

Discussion:

Patch:
diff --git a/.travis.yml b/.travis.yml
index 7296c15..46eb54f 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -49,8 +49,12 @@ before_script:
   - /bin/sh "$TRAVIS_BUILD_DIR/rabbitmq_server-$RABBITMQ_VERSION/sbin/rabbitmqctl" status
 
 script:
+  # See https://github.com/travis-ci/travis-ci/issues/1066 and https://github.com/pika/pika/pull/984#issuecomment-370565220
+  # as to why 'set -e' and 'set +e' are added here
+  - set -e
   - nosetests
   - PIKA_TEST_TLS=true nosetests
+  - set +e
 
 after_success:
   - aws s3 cp .coverage "s3://com-gavinroy-travis/pika/$TRAVIS_BUILD_NUMBER/.coverage.${TRAVIS_PYTHON_VERSION}"
diff --git a/docs/examples/tls_mutual_authentication.rst b/docs/examples/tls_mutual_authentication.rst
index 9cd8dec..8d0e164 100644
--- a/docs/examples/tls_mutual_authentication.rst
+++ b/docs/examples/tls_mutual_authentication.rst
@@ -18,14 +18,13 @@ tls_example.py::
 
     logging.basicConfig(level=logging.INFO)
 
-    cp = pika.ConnectionParameters(
-        ssl=True,
-        ssl_options=dict(
-            ssl_version=ssl.PROTOCOL_TLSv1,
-            ca_certs="/Users/me/tls-gen/basic/testca/cacert.pem",
-            keyfile="/Users/me/tls-gen/basic/client/key.pem",
-            certfile="/Users/me/tls-gen/basic/client/cert.pem",
-            cert_reqs=ssl.CERT_REQUIRED))
+    context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+    context.verify_mode = ssl.CERT_REQUIRED
+    context.load_verify_locations('/Users/me/tls-gen/basic/testca/cacert.pem')
+    context.load_cert_chain('/Users/me/tls-gen/basic/client/cert.pem',
+                            '/Users/me/tls-gen/basic/client/key.pem')
+
+    cp = pika.ConnectionParameters(ssl_options=pika.SSLOptions(context))
 
     conn = pika.BlockingConnection(cp)
     ch = conn.channel()
diff --git a/docs/examples/tls_server_uathentication.rst b/docs/examples/tls_server_authentication.rst
similarity index 89%
rename from docs/examples/tls_server_uathentication.rst
rename to docs/examples/tls_server_authentication.rst
index f2c56f1..79b58b9 100644
--- a/docs/examples/tls_server_uathentication.rst
+++ b/docs/examples/tls_server_authentication.rst
@@ -18,12 +18,11 @@ tls_example.py::
 
     logging.basicConfig(level=logging.INFO)
 
-    cp = pika.ConnectionParameters(
-        ssl=True,
-        ssl_options=dict(
-            ssl_version=ssl.PROTOCOL_TLSv1,
-            ca_certs="/Users/me/tls-gen/basic/testca/cacert.pem",
-            cert_reqs=ssl.CERT_REQUIRED))
+    context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+    context.verify_mode = ssl.CERT_REQUIRED
+    context.load_verify_locations('/Users/me/tls-gen/basic/testca/cacert.pem')
+
+    cp = pika.ConnectionParameters(ssl_options=pika.SSLOptions(context))
 
     conn = pika.BlockingConnection(cp)
     ch = conn.channel()
diff --git a/examples/twisted_service.py b/examples/twisted_service.py
index ced89a0..3057121 100644
--- a/examples/twisted_service.py
+++ b/examples/twisted_service.py
@@ -52,14 +52,14 @@ class PikaService(service.MultiService):
 
     def connect(self):
         f = PikaFactory(self.parameters)
-        if self.parameters.ssl:
+        if self.parameters.ssl_options:
             s = ssl.ClientContextFactory()
             serv = internet.SSLClient(host=self.parameters.host, port=self.parameters.port, factory=f, contextFactory=s)
         else:
             serv = internet.TCPClient(host=self.parameters.host, port=self.parameters.port, factory=f)
         serv.factory = f
         f.service = serv
-        name = '%s%s:%d' % ('ssl:' if self.parameters.ssl else '', self.parameters.host, self.parameters.port)
+        name = '%s%s:%d' % ('ssl:' if self.parameters.ssl_options else '', self.parameters.host, self.parameters.port)
         serv.__repr__ = lambda : '<AMQP Connection to %s>' % name
         serv.setName(name)
         serv.parent = self
diff --git a/pika/adapters/base_connection.py b/pika/adapters/base_connection.py
index 6a0db68..a79a577 100644
--- a/pika/adapters/base_connection.py
+++ b/pika/adapters/base_connection.py
@@ -57,9 +57,6 @@ class BaseConnection(connection.Connection):
             raise ValueError(
                 'Expected instance of Parameters, not %r' % parameters)
 
-        # Let the developer know we could not import SSL
-        if parameters and parameters.ssl and not ssl:
-            raise RuntimeError("SSL specified but it is not available")
         self.base_events = self.READ | self.ERROR
         self.event_state = self.base_events
         self.ioloop = ioloop
@@ -215,7 +212,7 @@ class BaseConnection(connection.Connection):
         pika.tcp_socket_opts.set_sock_opts(self.params.tcp_options, self.socket)
 
         # Wrap socket if using SSL
-        if self.params.ssl:
+        if self.params.ssl_options is not None:
             self.socket = self._wrap_socket(self.socket)
             ssl_text = " with SSL"
         else:
@@ -240,7 +237,7 @@ class BaseConnection(connection.Connection):
             return error
 
         # Handle SSL Connection Negotiation
-        if self.params.ssl and self.DO_HANDSHAKE:
+        if self.params.ssl_options is not None and self.DO_HANDSHAKE:
             try:
                 self._do_ssl_handshake()
             except ssl.SSLError as error:
@@ -322,7 +319,7 @@ class BaseConnection(connection.Connection):
         if exception.errno in self.ERRORS_TO_ABORT:
             LOGGER.error("Fatal Socket Error: %r", exception)
 
-        elif self.params.ssl and isinstance(exception, ssl.SSLError):
+        elif isinstance(exception, ssl.SSLError):
 
             if exception.errno == ssl.SSL_ERROR_WANT_READ:
                 # TODO doesn't seem right: this logic updates event state, but
@@ -384,7 +381,7 @@ class BaseConnection(connection.Connection):
         try:
             while True:
                 try:
-                    if self.params.ssl:
+                    if self.params.ssl_options is not None:
                         # TODO Why using read instead of recv on ssl socket?
                         data = self.socket.read(self._buffer_size)
                     else:
@@ -505,37 +502,11 @@ class BaseConnection(connection.Connection):
         :rtype: ssl.SSLSocket
 
         """
-        ssl_options = self.params.ssl_options or {}
-        # our wrapped return sock
-        ssl_sock = None
-
-        if isinstance(ssl_options, connection.SSLOptions):
-            context = ssl.SSLContext(ssl_options.ssl_version)
-            context.verify_mode = ssl_options.verify_mode
-            if ssl_options.certfile is not None:
-                context.load_cert_chain(
-                    certfile=ssl_options.certfile,
-                    keyfile=ssl_options.keyfile,
-                    password=ssl_options.key_password)
-
-            # only one of either cafile or capath have to defined
-            if ssl_options.cafile is not None or ssl_options.capath is not None:
-                context.load_verify_locations(
-                    cafile=ssl_options.cafile,
-                    capath=ssl_options.capath,
-                    cadata=ssl_options.cadata)
-
-            if ssl_options.ciphers is not None:
-                context.set_ciphers(ssl_options.ciphers)
-
-            ssl_sock = context.wrap_socket(
-                sock,
-                server_side=ssl_options.server_side,
-                do_handshake_on_connect=ssl_options.do_handshake_on_connect,
-                suppress_ragged_eofs=ssl_options.suppress_ragged_eofs,
-                server_hostname=ssl_options.server_hostname)
-        else:
-            ssl_sock = ssl.wrap_socket(
-                sock, do_handshake_on_connect=self.DO_HANDSHAKE, **ssl_options)
-
-        return ssl_sock
+        ssl_options = self.params.ssl_options
+
+        return ssl_options.context.wrap_socket(
+            sock,
+            server_side=False,
+            do_handshake_on_connect=self.DO_HANDSHAKE,
+            suppress_ragged_eofs=True,
+            server_hostname=ssl_options.server_hostname)
diff --git a/pika/compat.py b/pika/compat.py
index ad14aaa..c5b4ae2 100644
--- a/pika/compat.py
+++ b/pika/compat.py
@@ -23,7 +23,8 @@ except AttributeError:
 if not PY2:
     # these were moved around for Python 3
     from urllib.parse import (quote as url_quote, unquote as url_unquote,
-                              urlencode)
+                              urlencode, parse_qs as url_parse_qs,
+                              urlparse)
 
     # Python 3 does not have basestring anymore; we include
     # *only* the str here as this is used for textual data.
@@ -111,6 +112,8 @@ if not PY2:
 else:
     from urllib import quote as url_quote, unquote as url_unquote, urlencode
 
+    from urlparse import parse_qs as url_parse_qs, urlparse
+
     basestring = basestring
     str_or_bytes = basestring
     xrange = xrange
diff --git a/pika/connection.py b/pika/connection.py
index c784cc2..287bf81 100644
--- a/pika/connection.py
+++ b/pika/connection.py
@@ -3,21 +3,16 @@
 # pylint: disable=C0302
 
 import ast
-import sys
 import collections
 import copy
 import logging
 import math
 import numbers
 import platform
+import socket
 import warnings
 import ssl
 
-if sys.version_info > (3,):
-    import urllib.parse as urlparse  # pylint: disable=E0611,F0401
-else:
-    import urlparse
-
 from pika import __version__
 from pika import callback as pika_callback
 import pika.channel
@@ -28,6 +23,7 @@ from pika import heartbeat as pika_heartbeat
 
 from pika import spec
 
+import pika.compat
 from pika.compat import (xrange, basestring, # pylint: disable=W0622
                          url_unquote, dictkeys, dict_itervalues,
                          dict_iteritems)
@@ -55,23 +51,6 @@ class InternalCloseReasons(object):
 class Parameters(object):  # pylint: disable=R0902
     """Base connection parameters class definition
 
-    :param bool backpressure_detection: `DEFAULT_BACKPRESSURE_DETECTION`
-    :param float|None blocked_connection_timeout:
-        `DEFAULT_BLOCKED_CONNECTION_TIMEOUT`
-    :param int channel_max: `DEFAULT_CHANNEL_MAX`
-    :param int connection_attempts: `DEFAULT_CONNECTION_ATTEMPTS`
-    :param  credentials: `DEFAULT_CREDENTIALS`
-    :param int frame_max: `DEFAULT_FRAME_MAX`
-    :param int heartbeat: `DEFAULT_HEARTBEAT_TIMEOUT`
-    :param str host: `DEFAULT_HOST`
-    :param str locale: `DEFAULT_LOCALE`
-    :param int port: `DEFAULT_PORT`
-    :param float retry_delay: `DEFAULT_RETRY_DELAY`
-    :param float socket_timeout: `DEFAULT_SOCKET_TIMEOUT`
-    :param bool ssl: `DEFAULT_SSL`
-    :param dict ssl_options: `DEFAULT_SSL_OPTIONS`
-    :param str virtual_host: `DEFAULT_VIRTUAL_HOST`
-    :param int tcp_options: `DEFAULT_TCP_OPTIONS`
     """
 
     # Declare slots to protect against accidental assignment of an invalid
@@ -90,7 +69,6 @@ class Parameters(object):  # pylint: disable=R0902
         '_port',
         '_retry_delay',
         '_socket_timeout',
-        '_ssl',
         '_ssl_options',
         '_virtual_host',
         '_tcp_options'
@@ -166,9 +144,6 @@ class Parameters(object):  # pylint: disable=R0902
         self._socket_timeout = None
         self.socket_timeout = self.DEFAULT_SOCKET_TIMEOUT
 
-        self._ssl = None
-        self.ssl = self.DEFAULT_SSL
-
         self._ssl_options = None
         self.ssl_options = self.DEFAULT_SSL_OPTIONS
 
@@ -186,7 +161,7 @@ class Parameters(object):  # pylint: disable=R0902
         """
         return ('<%s host=%s port=%s virtual_host=%s ssl=%s>' %
                 (self.__class__.__name__, self.host, self.port,
-                 self.virtual_host, self.ssl))
+                 self.virtual_host, bool(self.ssl_options)))
 
     @property
     def backpressure_detection(self):
@@ -494,47 +469,26 @@ class Parameters(object):  # pylint: disable=R0902
                                  (value,))
         self._socket_timeout = value
 
-    @property
-    def ssl(self):
-        """
-        :returns: boolean indicating whether to connect via SSL. Defaults to
-            `DEFAULT_SSL`.
-
-        """
-        return self._ssl
-
-    @ssl.setter
-    def ssl(self, value):
-        """
-        :param bool value: boolean indicating whether to connect via SSL
-
-        """
-        if not isinstance(value, bool):
-            raise TypeError('ssl must be a bool, but got %r' % (value,))
-        self._ssl = value
-
     @property
     def ssl_options(self):
         """
-        :returns: None or a dict of options to pass to `ssl.wrap_socket`.
-            Defaults to `DEFAULT_SSL_OPTIONS`.
-
+        :returns: None for plaintext or `pika.SSLOptions` instance for SSL/TLS.
+        :rtype: `pika.SSLOptions`|None
         """
         return self._ssl_options
 
     @ssl_options.setter
     def ssl_options(self, value):
         """
-        :param value: None, a dict of options to pass to `ssl.wrap_socket` or
-            a SSLOptions object for advanced setup.
+        :param `pika.SSLOptions`|None value: None for plaintext or
+            `pika.SSLOptions` instance for SSL/TLS. Defaults to None.
 
         """
-        if not isinstance(value, (dict, SSLOptions, type(None))):
+        if not isinstance(value, (SSLOptions, type(None))):
             raise TypeError(
-                'ssl_options must be a dict, None or an SSLOptions but got %r'
+                'ssl_options must be None or SSLOptions but got %r'
                 % (value, ))
-        # Copy the mutable object to avoid accidental side-effects
-        self._ssl_options = copy.deepcopy(value)
+        self._ssl_options = value
 
 
     @property
@@ -560,13 +514,14 @@ class Parameters(object):  # pylint: disable=R0902
     def tcp_options(self):
         """
         :returns: None or a dict of options to pass to the underlying socket
+        :rtype: dict|None
         """
         return self._tcp_options
 
     @tcp_options.setter
     def tcp_options(self, value):
         """
-        :param bool value: None or a dict of options to pass to the underlying
+        :param dict|None value: None or a dict of options to pass to the underlying
             socket. Currently supported are TCP_KEEPIDLE, TCP_KEEPINTVL, TCP_KEEPCNT
             and TCP_USER_TIMEOUT. Availability of these may depend on your platform.
         """
@@ -597,7 +552,6 @@ class ConnectionParameters(Parameters):
                  channel_max=_DEFAULT,
                  frame_max=_DEFAULT,
                  heartbeat=_DEFAULT,
-                 ssl=_DEFAULT,
                  ssl_options=_DEFAULT,
                  connection_attempts=_DEFAULT,
                  retry_delay=_DEFAULT,
@@ -624,9 +578,8 @@ class ConnectionParameters(Parameters):
             with the connection instance and the heartbeat timeout proposed by broker
             as its arguments. The callback should return a non-negative integer that
             will be used to override the broker's proposal.
-        :param bool ssl: Enable SSL
-        :param dict ssl_options: None or a dict of arguments to be passed to
-            ssl.wrap_socket
+        :param `pika.SSLOptions`|None ssl_options: None for plaintext or
+            `pika.SSLOptions` instance for SSL/TLS. Defaults to None.
         :param int connection_attempts: Maximum number of retry attempts
         :param int|float retry_delay: Time to wait in seconds, before the next
         :param int|float socket_timeout: Use for high latency networks
@@ -701,17 +654,14 @@ class ConnectionParameters(Parameters):
         if socket_timeout is not self._DEFAULT:
             self.socket_timeout = socket_timeout
 
-        if ssl is not self._DEFAULT:
-            self.ssl = ssl
-
         if ssl_options is not self._DEFAULT:
             self.ssl_options = ssl_options
 
         # Set port after SSL status is known
         if port is not self._DEFAULT:
             self.port = port
-        elif ssl is not self._DEFAULT:
-            self.port = self.DEFAULT_SSL_PORT if self.ssl else self.DEFAULT_PORT
+        else:
+            self.port = self.DEFAULT_SSL_PORT if self.ssl_options else self.DEFAULT_PORT
 
         if virtual_host is not self._DEFAULT:
             self.virtual_host = virtual_host
@@ -754,10 +704,11 @@ class URLParameters(Parameters):
             the broker's value is accepted. 0 turns heartbeat off.
         - locale:
             Override the default `en_US` locale value
-        - ssl:
-            Toggle SSL, possible values are `t`, `f`
         - ssl_options:
-            Arguments passed to :meth:`ssl.wrap_socket`
+            None for plaintext; for SSL: dict of public ssl context-related
+            arguments that may be passed to :meth:`ssl.SSLSocket` as kwargs,
+            except `sock`, `server_side`,`do_handshake_on_connect`, `family`,
+            `type`, `proto`, `fileno`.
         - retry_delay:
             The number of seconds to sleep before attempting to connect on
             connection failure.
@@ -804,12 +755,15 @@ class URLParameters(Parameters):
 
         # TODO Is support for the alternative http(s) schemes intentional?
 
-        parts = urlparse.urlparse(url)
+        parts = pika.compat.urlparse(url)
 
         if parts.scheme == 'https':
-            self.ssl = True
+            # Create default context which will get overridden by the
+            # ssl_options URL arg, if any
+            self.ssl_options = pika.SSLOptions(
+                context=ssl.create_default_context())
         elif parts.scheme == 'http':
-            self.ssl = False
+            self.ssl_options = None
         elif parts.scheme:
             raise ValueError('Unexpected URL scheme %r; supported scheme '
                              'values: amqp, amqps' % (parts.scheme,))
@@ -821,7 +775,8 @@ class URLParameters(Parameters):
         if parts.port is not None:
             self.port = parts.port
         else:
-            self.port = self.DEFAULT_SSL_PORT if self.ssl else self.DEFAULT_PORT
+            self.port = (self.DEFAULT_SSL_PORT if self.ssl_options
+                         else self.DEFAULT_PORT)
 
         if parts.username is not None:
             self.credentials = pika_credentials.PlainCredentials(url_unquote(parts.username),
@@ -833,7 +788,7 @@ class URLParameters(Parameters):
 
         # Handle query string values, validating and assigning them
 
-        self._all_url_query_values = urlparse.parse_qs(parts.query)
+        self._all_url_query_values = pika.compat.url_parse_qs(parts.query)
 
         for name, value in dict_iteritems(self._all_url_query_values):
             try:
@@ -947,66 +902,60 @@ class URLParameters(Parameters):
         self.socket_timeout = socket_timeout
 
     def _set_url_ssl_options(self, value):
-        """Deserialize and apply the corresponding query string arg"""
-        self.ssl_options = ast.literal_eval(value)
+        """Deserialize and apply the corresponding query string arg
+
+        """
+        options = ast.literal_eval(value)
+        if options is None:
+            if self.ssl_options is not None:
+                raise ValueError(
+                    'Specified ssl_options=None URL arg is inconsistent with '
+                    'the specified https URL scheme.')
+        else:
+            # Convert options to pika.SSLOptions via ssl.SSLSocket()
+            sock = socket.socket()
+            try:
+                ssl_sock = ssl.SSLSocket(sock=sock, **options)
+                try:
+                    self.ssl_options = pika.SSLOptions(
+                        context=ssl_sock.context,
+                        server_hostname=ssl_sock.server_hostname)
+                finally:
+                    ssl_sock.close()
+            finally:
+                sock.close()
+
+
 
     def _set_url_tcp_options(self, value):
         """Deserialize and apply the corresponding query string arg"""
         self.tcp_options = ast.literal_eval(value)
 
+
 class SSLOptions(object):
     """Class used to provide parameters for optional fine grained control of SSL
     socket wrapping.
 
-    :param string keyfile: The key file to pass to SSLContext.load_cert_chain
-    :param string key_password: The key password to passed to
-                                                    SSLContext.load_cert_chain
-    :param string certfile: The certificate file to passed to
-                                                    SSLContext.load_cert_chain
-    :param bool server_side: Passed to SSLContext.wrap_socket
-    :param verify_mode: Passed to SSLContext.wrap_socket
-    :param ssl_version: Passed to SSLContext init, defines the ssl
-                                                                version to use
-    :param string cafile: The CA file passed to
-                                            SSLContext.load_verify_locations
-    :param string capath: The CA path passed to
-                                            SSLContext.load_verify_locations
-    :param string cadata: The CA data passed to
-                                            SSLContext.load_verify_locations
-    :param do_handshake_on_connect: Passed to SSLContext.wrap_socket
-    :param suppress_ragged_eofs: Passed to SSLContext.wrap_socket
-    :param ciphers: Passed to SSLContext.set_ciphers
-    :param server_hostname: SSLContext.wrap_socket, used to enable SNI
     """
 
-    def __init__(self,
-                 keyfile=None,
-                 key_password=None,
-                 certfile=None,
-                 server_side=False,
-                 verify_mode=ssl.CERT_NONE,
-                 ssl_version=ssl.PROTOCOL_SSLv23,
-                 cafile=None,
-                 capath=None,
-                 cadata=None,
-                 do_handshake_on_connect=True,
-                 suppress_ragged_eofs=True,
-                 ciphers=None,
-                 server_hostname=None):
-        self.keyfile = keyfile
-        self.key_password = key_password
-        self.certfile = certfile
-        self.server_side = server_side
-        self.verify_mode = verify_mode
-        self.ssl_version = ssl_version
-        self.cafile = cafile
-        self.capath = capath
-        self.cadata = cadata
-        self.do_handshake_on_connect = do_handshake_on_connect
-        self.suppress_ragged_eofs = suppress_ragged_eofs
-        self.ciphers = ciphers
+    # Protect against accidental assignment of an invalid attribute
+    __slots__ = ('context', 'server_hostname')
+
+    def __init__(self, context, server_hostname=None):
+        """
+        :param ssl.SSLContext context: SSLContext instance
+        :param str|None server_hostname: SSLContext.wrap_socket, used to enable
+            SNI
+        """
+        if not isinstance(context, ssl.SSLContext):
+            raise TypeError(
+                'context must be of ssl.SSLContext type, but got {!r}'.format(
+                    context))
+
+        self.context = context
         self.server_hostname = server_hostname
 
+
 class Connection(object):
     """This is the core class that implements communication with RabbitMQ. This
     class should not be invoked directly but rather through the use of an
@@ -1076,8 +1025,18 @@ class Connection(object):
         self.heartbeat = None
 
         # Set our configuration options
-        self.params = (copy.deepcopy(parameters) if parameters is not None else
-                       ConnectionParameters())
+        if parameters is not None:
+            # NOTE: Work around inability to copy ssl.SSLContext contained in
+            # our SSLOptions; ssl.SSLContext fails to implement __getnewargs__
+            saved_ssl_options = parameters.ssl_options
+            parameters.ssl_options = None
+            try:
+                self.params = copy.deepcopy(parameters)
+                self.params.ssl_options = saved_ssl_options
+            finally:
+                parameters.ssl_options = saved_ssl_options
+        else:
+            self.params = ConnectionParameters()
 
         # Define our callback dictionary
         self.callbacks = pika_callback.CallbackManager()


Test patch:
diff --git a/tests/acceptance/async_test_base.py b/tests/acceptance/async_test_base.py
index 1df0eb0..caf237d 100644
--- a/tests/acceptance/async_test_base.py
+++ b/tests/acceptance/async_test_base.py
@@ -37,13 +37,12 @@ class AsyncTestCase(unittest.TestCase):
         if self.should_test_tls():
             self.logger.info('testing using TLS/SSL connection to port 5671')
             self.parameters.port = 5671
-            self.parameters.ssl = True
-            self.parameters.ssl_options = dict(
-                ssl_version=ssl.PROTOCOL_TLSv1,
-                ca_certs="testdata/certs/ca_certificate.pem",
-                keyfile="testdata/certs/client_key.pem",
-                certfile="testdata/certs/client_certificate.pem",
-                cert_reqs=ssl.CERT_REQUIRED)
+            context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+            context.verify_mode = ssl.CERT_REQUIRED
+            context.load_verify_locations('testdata/certs/ca_certificate.pem')
+            context.load_cert_chain('testdata/certs/client_certificate.pem',
+                                    'testdata/certs/client_key.pem')
+            self.parameters.ssl_options = pika.SSLOptions(context)
         self._timed_out = False
         super(AsyncTestCase, self).setUp()
 
diff --git a/tests/unit/base_connection_tests.py b/tests/unit/base_connection_tests.py
index d35b3a2..6c23500 100644
--- a/tests/unit/base_connection_tests.py
+++ b/tests/unit/base_connection_tests.py
@@ -4,13 +4,13 @@ Tests for pika.base_connection.BaseConnection
 """
 
 import socket
+import ssl
+import sys
 import unittest
 
 import mock
 
 import pika
-import sys
-import ssl
 
 import pika.tcp_socket_opts
 from pika.adapters import base_connection
@@ -46,9 +46,8 @@ class BaseConnectionTests(unittest.TestCase):
         params = pika.ConnectionParameters(tcp_options=tcp_options)
         self.assertEqual(params.tcp_options, tcp_options)
 
-        with mock.patch.dict('pika.tcp_socket_opts._SUPPORTED_TCP_OPTIONS', {
-                'TCP_KEEPIDLE': socket.TCP_KEEPIDLE
-        }):
+        with mock.patch.dict('pika.tcp_socket_opts._SUPPORTED_TCP_OPTIONS',
+                             {'TCP_KEEPIDLE': socket.TCP_KEEPIDLE}):
             sock_mock = mock.Mock()
             pika.tcp_socket_opts.set_sock_opts(params.tcp_options, sock_mock)
 
@@ -84,56 +83,13 @@ class BaseConnectionTests(unittest.TestCase):
                                               socket.SO_KEEPALIVE, 1)
         self.assertNotIn(keepalive_call, sock_mock.method_calls)
 
-    def test_ssl_wrap_socket_with_none_ssl_options(self):
-
-        params = pika.ConnectionParameters(ssl_options=None)
-        self.assertIsNone(params.ssl_options)
-
-        with mock.patch('pika.connection.Connection.connect'):
-            conn = base_connection.BaseConnection(parameters=params)
-
-            with mock.patch('pika.adapters.base_connection'
-                            '.ssl.wrap_socket') as wrap_socket_mock:
-                sock_mock = mock.Mock()
-                conn._wrap_socket(sock_mock)
-
-                wrap_socket_mock.assert_called_once_with(
-                    sock_mock, do_handshake_on_connect=conn.DO_HANDSHAKE)
-
-    def test_ssl_wrap_socket_with_dict_ssl_options(self):
-
-        ssl_options = dict(ssl='options', handshake=False)
-        params = pika.ConnectionParameters(ssl_options=ssl_options)
-        self.assertEqual(params.ssl_options, ssl_options)
-
-        with mock.patch('pika.connection.Connection.connect'):
-            conn = base_connection.BaseConnection(parameters=params)
-
-            with mock.patch('pika.adapters.base_connection'
-                            '.ssl.wrap_socket') as wrap_socket_mock:
-                sock_mock = mock.Mock()
-                conn._wrap_socket(sock_mock)
-
-                wrap_socket_mock.assert_called_once_with(
-                    sock_mock,
-                    do_handshake_on_connect=conn.DO_HANDSHAKE,
-                    ssl='options',
-                    handshake=False)
-
-    @mock.patch('ssl.SSLContext.load_cert_chain')
-    @mock.patch('ssl.SSLContext.load_verify_locations')
-    @mock.patch('ssl.SSLContext.set_ciphers')
     @mock.patch('ssl.SSLContext.wrap_socket')
-    @unittest.skipIf(sys.version_info < (2,7,0), 'Unavailable ssl features')
+    @unittest.skipIf(sys.version_info < (2, 7, 0), 'Unavailable ssl features')
     def test_ssl_wrap_socket_with_default_ssl_options_obj(self,
-                                                          wrap_socket_mock,
-                                                          set_ciphers_mock,
-                                                          load_verify_mock,
-                                                          load_certs_mock):
-        ssl_options = pika.SSLOptions()
+                                                          wrap_socket_mock):
+        ssl_options = pika.SSLOptions(context=ssl.create_default_context())
         params = pika.ConnectionParameters(ssl_options=ssl_options)
-        #self.assertEqual(params.ssl_options, ssl_options)
-
+        self.assertIs(params.ssl_options, ssl_options)
 
         with mock.patch('pika.connection.Connection.connect'):
             conn = base_connection.BaseConnection(parameters=params)
@@ -141,40 +97,18 @@ class BaseConnectionTests(unittest.TestCase):
             sock_mock = mock.Mock()
             conn._wrap_socket(sock_mock)
 
-            load_certs_mock.assert_not_called()
-            load_verify_mock.assert_not_called()
-            # the __init__ of SSLContext calls set_ciphers,
-            # hence the 'called once'
-            set_ciphers_mock.assert_called_once()
             wrap_socket_mock.assert_called_once_with(
                 sock_mock,
                 server_side=False,
                 do_handshake_on_connect=conn.DO_HANDSHAKE,
                 suppress_ragged_eofs=True,
-                server_hostname=None
-                )
+                server_hostname=None)
 
-    @mock.patch('ssl.SSLContext.load_cert_chain')
-    @mock.patch('ssl.SSLContext.load_verify_locations')
-    @mock.patch('ssl.SSLContext.set_ciphers')
     @mock.patch('ssl.SSLContext.wrap_socket')
-    @unittest.skipIf(sys.version_info < (2,7,0), 'Unavailable ssl features')
-    def test_ssl_wrap_socket_with_ssl_options_obj(self,
-                                                        wrap_socket_mock,
-                                                        set_ciphers_mock,
-                                                        load_verify_mock,
-                                                        load_certs_mock):
-        ssl_options = pika.SSLOptions(certfile='/some/cert/file.crt',
-                                      keyfile='/some/key/file.crt',
-                                      key_password='pa55w0rd',
-                                      cafile='/some/ca/file.crt',
-                                      capath='/some/ca/path',
-                                      cadata='/some/data/or/something',
-                                      ciphers='ciphers',
-                                      server_hostname='some.virtual.host',
-                                      do_handshake_on_connect=False,
-                                      suppress_ragged_eofs=False,
-                                      server_side=True)
+    @unittest.skipIf(sys.version_info < (2, 7, 0), 'Unavailable ssl features')
+    def test_ssl_wrap_socket_with_ssl_options_obj(self, wrap_socket_mock):
+        ssl_options = pika.SSLOptions(context=ssl.create_default_context(),
+                                      server_hostname='some.virtual.host')
         params = pika.ConnectionParameters(ssl_options=ssl_options)
         #self.assertEqual(params.ssl_options, ssl_options)
 
@@ -185,18 +119,9 @@ class BaseConnectionTests(unittest.TestCase):
             sock_mock = mock.Mock()
             conn._wrap_socket(sock_mock)
 
-            load_certs_mock.assert_called_once_with(certfile='/some/cert/file.crt',
-                                                    keyfile='/some/key/file.crt',
-                                                    password='pa55w0rd')
-            load_verify_mock.assert_called_once_with(cafile='/some/ca/file.crt',
-                                                     capath='/some/ca/path',
-                                                     cadata='/some/data/or/something')
-            # the constructor of SSLContext calls set_ciphers as well
-            set_ciphers_mock.assert_called_with('ciphers')
             wrap_socket_mock.assert_called_once_with(
                 sock_mock,
-                server_side=True,
-                do_handshake_on_connect=False,
-                suppress_ragged_eofs=False,
-                server_hostname='some.virtual.host'
-                )
+                server_side=False,
+                do_handshake_on_connect=True,
+                suppress_ragged_eofs=True,
+                server_hostname='some.virtual.host')
diff --git a/tests/unit/connection_parameters_tests.py b/tests/unit/connection_parameters_tests.py
index b71c8c3..159a54a 100644
--- a/tests/unit/connection_parameters_tests.py
+++ b/tests/unit/connection_parameters_tests.py
@@ -4,6 +4,7 @@ Test `pika.connection.Parameters`, `pika.connection.ConnectionParameters`, and
 
 """
 import copy
+import ssl
 import unittest
 import warnings
 
@@ -65,8 +66,6 @@ class _ParametersTestsBase(unittest.TestCase):
             kls.DEFAULT_RETRY_DELAY,
             'socket_timeout':
             kls.DEFAULT_SOCKET_TIMEOUT,
-            'ssl':
-            kls.DEFAULT_SSL,
             'ssl_options':
             kls.DEFAULT_SSL_OPTIONS,
             'virtual_host':
@@ -345,36 +344,18 @@ class ParametersTests(_ParametersTestsBase):
         with self.assertRaises(ValueError):
             params.socket_timeout = 0
 
-    def test_ssl(self):
-        params = connection.Parameters()
-
-        params.ssl = False
-        self.assertEqual(params.ssl, False)
-
-        params.ssl = True
-        self.assertEqual(params.ssl, True)
-
-        with self.assertRaises(TypeError):
-            params.backpressure_detection = 1
-
-        with self.assertRaises(TypeError):
-            params.ssl = 'True'
-
-        with self.assertRaises(TypeError):
-            params.ssl = 'f'
-
     def test_ssl_options(self):
         params = connection.Parameters()
 
-        opt = dict(key='value', key2=2, key3=dict(a=1))
-        params.ssl_options = copy.deepcopy(opt)
-        self.assertEqual(params.ssl_options, opt)
+        ssl_options = connection.SSLOptions(ssl.create_default_context())
+        params.ssl_options = ssl_options
+        self.assertIs(params.ssl_options, ssl_options)
 
         params.ssl_options = None
         self.assertIsNone(params.ssl_options)
 
         with self.assertRaises(TypeError):
-            params.ssl_options = str(opt)
+            params.ssl_options = dict()
 
     def test_virtual_host(self):
         params = connection.Parameters()
@@ -414,27 +395,30 @@ class ConnectionParametersTests(_ParametersTestsBase):
         self.assert_default_parameter_values(connection.ConnectionParameters())
 
     def test_explicit_ssl_with_default_port(self):
-        params = connection.ConnectionParameters(ssl=True)
+        params = connection.ConnectionParameters(
+            ssl_options=connection.SSLOptions(ssl.create_default_context()))
 
-        self.assertEqual(params.ssl, True)
+        self.assertIsNotNone(params.ssl_options)
         self.assertEqual(params.port, params.DEFAULT_SSL_PORT)
 
     def test_explicit_ssl_with_explict_port(self):
-        params = connection.ConnectionParameters(ssl=True, port=99)
+        params = connection.ConnectionParameters(
+            ssl_options=connection.SSLOptions(ssl.create_default_context()),
+            port=99)
 
-        self.assertEqual(params.ssl, True)
+        self.assertIsNotNone(params.ssl_options)
         self.assertEqual(params.port, 99)
 
     def test_explicit_non_ssl_with_default_port(self):
-        params = connection.ConnectionParameters(ssl=False)
+        params = connection.ConnectionParameters(ssl_options=None)
 
-        self.assertEqual(params.ssl, False)
+        self.assertIsNone(params.ssl_options)
         self.assertEqual(params.port, params.DEFAULT_PORT)
 
     def test_explicit_non_ssl_with_explict_port(self):
-        params = connection.ConnectionParameters(ssl=False, port=100)
+        params = connection.ConnectionParameters(ssl_options=None, port=100)
 
-        self.assertEqual(params.ssl, False)
+        self.assertIsNone(params.ssl_options)
         self.assertEqual(params.port, 100)
 
     def test_good_connection_parameters(self):
@@ -455,10 +439,7 @@ class ConnectionParametersTests(_ParametersTestsBase):
             'port': 5678,
             'retry_delay': 3,
             'socket_timeout': 100.5,
-            'ssl': True,
-            'ssl_options': {
-                'ssl': 'options'
-            },
+            'ssl_options': None,
             'virtual_host': u'vvhost',
             'tcp_options': {
                 'TCP_USER_TIMEOUT': 1000
@@ -560,38 +541,38 @@ class URLParametersTests(_ParametersTestsBase):
         params = connection.URLParameters('')
         self.assert_default_parameter_values(params)
 
-        self.assertEqual(params.ssl, False)
+        self.assertIsNone(params.ssl_options)
         self.assertEqual(params.port, params.DEFAULT_PORT)
 
     def test_no_ssl(self):
         params = connection.URLParameters('http://')
-        self.assertEqual(params.ssl, False)
+        self.assertIsNone(params.ssl_options)
         self.assertEqual(params.port, params.DEFAULT_PORT)
         self.assert_default_parameter_values(params)
 
         params = connection.URLParameters('amqp://')
-        self.assertEqual(params.ssl, False)
+        self.assertIsNone(params.ssl_options)
         self.assertEqual(params.port, params.DEFAULT_PORT)
         self.assert_default_parameter_values(params)
 
     def test_ssl(self):
         params = connection.URLParameters('https://')
-        self.assertEqual(params.ssl, True)
+        self.assertIsNotNone(params.ssl_options)
         self.assertEqual(params.port, params.DEFAULT_SSL_PORT)
 
         params = connection.URLParameters('amqps://')
-        self.assertEqual(params.ssl, True)
+        self.assertIsNotNone(params.ssl_options)
         self.assertEqual(params.port, params.DEFAULT_SSL_PORT)
 
         # Make sure the other parameters unrelated to SSL have default values
         params = connection.URLParameters('amqps://')
-        params.ssl = False
+        params.ssl_options = None
         params.port = params.DEFAULT_PORT
         self.assert_default_parameter_values(params)
 
     def test_no_url_scheme_defaults_to_plaintext(self):
         params = connection.URLParameters('//')
-        self.assertEqual(params.ssl, False)
+        self.assertIsNone(params.ssl_options)
         self.assertEqual(params.port, params.DEFAULT_PORT)
 
     def test_good_parameters(self):
@@ -605,8 +586,19 @@ class URLParametersTests(_ParametersTestsBase):
             'locale': 'en_UK',
             'retry_delay': 3,
             'socket_timeout': 100.5,
+            # NOTE: just straight ssl.CERT_OPTIONAL breaks on python 3.6 and 3.7
+            # during ast.literal_eval() of the urlencoded dict as invalid syntax
+            # on <VerifyMode.CERT_NONE: 1>:
+            # {'cert_reqs': <VerifyMode.CERT_NONE: 1>, 'server_hostname': 'blah.blah.com'}
             'ssl_options': {
-                'ssl': 'options'
+                'keyfile': None,
+                'certfile': None,
+                'ssl_version': int(ssl.PROTOCOL_SSLv23),
+                'ca_certs': None,
+                'cert_reqs': int(ssl.CERT_NONE),
+                'npn_protocols': None,
+                'ciphers': None,
+                'server_hostname': 'blah.blah.com'
             },
             'tcp_options': {
                 'TCP_USER_TIMEOUT': 1000,
@@ -615,7 +607,7 @@ class URLParametersTests(_ParametersTestsBase):
         }
 
         for backpressure in ('t', 'f'):
-            test_params = copy.deepcopy(query_args)
+            test_params = dict(query_args)
             test_params['backpressure_detection'] = backpressure
             virtual_host = '/'
             query_string = urlencode(test_params)
@@ -632,17 +624,23 @@ class URLParametersTests(_ParametersTestsBase):
                 expected_value = query_args[t_param]
                 actual_value = getattr(params, t_param)
 
-                self.assertEqual(
-                    actual_value,
-                    expected_value,
-                    msg='Expected %s=%r, but got %r' %
-                    (t_param, expected_value, actual_value))
+                if t_param == 'ssl_options':
+                    self.assertEqual(actual_value.context.verify_mode,
+                                     expected_value['cert_reqs'])
+                    self.assertEqual(actual_value.server_hostname,
+                                     expected_value['server_hostname'])
+                else:
+                    self.assertEqual(
+                        actual_value,
+                        expected_value,
+                        msg='Expected %s=%r, but got %r' %
+                        (t_param, expected_value, actual_value))
 
             self.assertEqual(params.backpressure_detection,
                              backpressure == 't')
 
             # check all values from base URL
-            self.assertEqual(params.ssl, True)
+            self.assertIsNotNone(params.ssl_options)
             self.assertEqual(params.credentials.username, 'myuser')
             self.assertEqual(params.credentials.password, 'mypass')
             self.assertEqual(params.host, 'www.test.com')