feat(core): Added SSL support (#513)

* client: Allow SSL use when communicating with Zookeeper, fixes #382

Zookeeper 3.5 supports SSL for client communications, this commit
adds support for it on the Kazoo side.

Note that you need to give the client the key, certificate and CA
files.

Co-Authored-By: Monty Taylor <mordred@inaugust.com>

* Added keyfile password for ssl connection

* Added a way to bypass ssl certification validation

* Added a timeout when using SSL connection

3 files changed, 98 insertions, 23 deletions

diff --git a/kazoo/client.py b/kazoo/client.py
index f583c06..1b0ff30 100644
--- a/kazoo/client.py
+++ b/kazoo/client.py
@@ -106,7 +106,9 @@ class KazooClient(object):
                  timeout=10.0, client_id=None, handler=None,
                  default_acl=None, auth_data=None, read_only=None,
                  randomize_hosts=True, connection_retry=None,
-                 command_retry=None, logger=None, **kwargs):
+                 command_retry=None, logger=None, keyfile=None,
+                 keyfile_password=None, certfile=None, ca=None,
+                 use_ssl=False, verify_certs=True, **kwargs):
         """Create a :class:`KazooClient` instance. All time arguments
         are in seconds.
 
@@ -135,6 +137,13 @@ class KazooClient(object):
             options which will be used for creating one.
         :param logger: A custom logger to use instead of the module
             global `log` instance.
+        :param keyfile: SSL keyfile to use for authentication
+        :param keyfile_password: SSL keyfile password
+        :param certfile: SSL certfile to use for authentication
+        :param ca: SSL CA file to use for authentication
+        :param use_ssl: argument to control whether SSL is used or not
+        :param verify_certs: when using SSL, argument to bypass
+            certs verification
 
         Basic Example:
 
@@ -183,6 +192,12 @@ class KazooClient(object):
         self.chroot = None
         self.set_hosts(hosts)
 
+        self.use_ssl = use_ssl
+        self.verify_certs = verify_certs
+        self.certfile = certfile
+        self.keyfile = keyfile
+        self.keyfile_password = keyfile_password
+        self.ca = ca
         # Curator like simplified state tracking, and listeners for
         # state transitions
         self._state = KeeperState.CLOSED
@@ -648,7 +663,14 @@ class KazooClient(object):
 
         peer = self._connection._socket.getpeername()[:2]
         sock = self.handler.create_connection(
-            peer, timeout=self._session_timeout / 1000.0)
+            peer, timeout=self._session_timeout / 1000.0,
+            use_ssl=self.use_ssl,
+            ca=self.ca,
+            certfile=self.certfile,
+            keyfile=self.keyfile,
+            keyfile_password=self.keyfile_password,
+            verify_certs=self.verify_certs,
+        )
         sock.sendall(cmd)
         result = sock.recv(8192)
         sock.close()
diff --git a/kazoo/handlers/utils.py b/kazoo/handlers/utils.py
index f18f3ac..0d36506 100644
--- a/kazoo/handlers/utils.py
+++ b/kazoo/handlers/utils.py
@@ -3,6 +3,8 @@
 import errno
 import functools
 import select
+import ssl
+import socket
 import time
 
 HAS_FNCTL = True
@@ -183,7 +185,10 @@ def create_tcp_socket(module):
     return sock
 
 
-def create_tcp_connection(module, address, timeout=None):
+def create_tcp_connection(module, address, timeout=None,
+                          use_ssl=False, ca=None, certfile=None,
+                          keyfile=None, keyfile_password=None,
+                          verify_certs=True):
     end = None
     if timeout is None:
         # thanks to create_connection() developers for
@@ -194,17 +199,48 @@ def create_tcp_connection(module, address, timeout=None):
     sock = None
 
     while end is None or time.time() < end:
-        try:
-            # if we got a timeout, lets ensure that we decrement the time
-            # otherwise there is no timeout set and we'll call it as such
-            timeout_at = end if end is None else end - time.time()
-            sock = module.create_connection(address, timeout_at)
-            break
-        except Exception as ex:
-            errnum = ex.errno if isinstance(ex, OSError) else ex[0]
-            if errnum == errno.EINTR:
-                continue
-            raise
+        timeout_at = end if end is None else end - time.time()
+        if use_ssl:
+            # Disallow use of SSLv2 and V3 (meaning we require TLSv1.0+)
+            context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+            context.options |= ssl.OP_NO_SSLv2
+            context.options |= ssl.OP_NO_SSLv3
+            # Load default CA certs
+            context.load_default_certs(ssl.Purpose.SERVER_AUTH)
+            context.verify_mode = (
+                ssl.CERT_OPTIONAL if verify_certs else ssl.CERT_NONE
+            )
+            if ca:
+                context.load_verify_locations(ca)
+            if certfile and keyfile:
+                context.verify_mode = (
+                    ssl.CERT_REQUIRED if verify_certs else ssl.CERT_NONE
+                )
+                context.load_cert_chain(certfile=certfile,
+                                        keyfile=keyfile,
+                                        password=keyfile_password)
+            try:
+                # Query the address to get back it's address family
+                addrs = socket.getaddrinfo(address[0], address[1], 0,
+                                           socket.SOCK_STREAM)
+                conn = context.wrap_socket(module.socket(addrs[0][0]))
+                conn.settimeout(timeout_at)
+                conn.connect(address)
+                sock = conn
+                break
+            except ssl.SSLError:
+                raise
+        else:
+            try:
+                # if we got a timeout, lets ensure that we decrement the time
+                # otherwise there is no timeout set and we'll call it as such
+                sock = module.create_connection(address, timeout_at)
+                break
+            except Exception as ex:
+                errnum = ex.errno if isinstance(ex, OSError) else ex[0]
+                if errnum == errno.EINTR:
+                    continue
+                raise
 
     if sock is None:
         raise module.error
diff --git a/kazoo/protocol/connection.py b/kazoo/protocol/connection.py
index d37f7c7..32c7398 100644
--- a/kazoo/protocol/connection.py
+++ b/kazoo/protocol/connection.py
@@ -216,13 +216,21 @@ class ConnectionHandler(object):
         remaining = length
         with self._socket_error_handling():
             while remaining > 0:
-                s = self.handler.select([self._socket], [], [], timeout)[0]
-                if not s:  # pragma: nocover
-                    # If the read list is empty, we got a timeout. We don't
-                    # have to check wlist and xlist as we don't set any
-                    raise self.handler.timeout_exception("socket time-out"
-                                                         " during read")
-
+                # Because of SSL framing, a select may not return when using
+                # an SSL socket because the underlying physical socket may not
+                # have anything to select, but the wrapped object may still
+                # have something to read as it has previously gotten enough
+                # data from the underlying socket.
+                if (hasattr(self._socket, "pending")
+                        and self._socket.pending() > 0):
+                    pass
+                else:
+                    s = self.handler.select([self._socket], [], [], timeout)[0]
+                    if not s:  # pragma: nocover
+                        # If the read list is empty, we got a timeout. We don't
+                        # have to check wlist and xlist as we don't set any
+                        raise self.handler.timeout_exception(
+                            "socket time-out during read")
                 chunk = self._socket.recv(remaining)
                 if chunk == b'':
                     raise ConnectionDropped('socket connection broken')
@@ -596,7 +604,8 @@ class ConnectionHandler(object):
 
     def _connect(self, host, port):
         client = self.client
-        self.logger.info('Connecting to %s:%s', host, port)
+        self.logger.info('Connecting to %s:%s, use_ssl: %r',
+                         host, port, self.client.use_ssl)
 
         self.logger.log(BLATHER,
                         '    Using session_id: %r session_passwd: %s',
@@ -605,7 +614,15 @@ class ConnectionHandler(object):
 
         with self._socket_error_handling():
             self._socket = self.handler.create_connection(
-                (host, port), client._session_timeout / 1000.0)
+                address=(host, port),
+                timeout=client._session_timeout / 1000.0,
+                use_ssl=self.client.use_ssl,
+                keyfile=self.client.keyfile,
+                certfile=self.client.certfile,
+                ca=self.client.ca,
+                keyfile_password=self.client.keyfile_password,
+                verify_certs=self.client.verify_certs,
+            )
 
         self._socket.setblocking(0)