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

Author: StephenSorriaux

kazoo/client.py

@@ -106,7 +106,9 @@ def __init__(self, hosts='127.0.0.1:2181',
                  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 @@ def __init__(self, hosts='127.0.0.1:2181',
             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 @@ def __init__(self, hosts='127.0.0.1:2181',
         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 @@ def command(self, cmd=b'ruok'):
 
         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()

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

kazoo/protocol/connection.py

@@ -216,13 +216,21 @@ def _read(self, length, timeout):
         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 @@ def _connect_attempt(self, host, port, retry):
 
     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 @@ def _connect(self, host, port):
 
         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)