Add CRL support to SSL support (#683)

A user can provide a CRL whose peer certificate will be checked
against. This only works with Python 3.4+ and Python 2.7.9+.

4 files changed, 29 insertions, 0 deletions

diff --git a/kafka/client_async.py b/kafka/client_async.py
index 7719426..7079f01 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -59,6 +59,7 @@ class KafkaClient(object):
         'ssl_cafile': None,
         'ssl_certfile': None,
         'ssl_keyfile': None,
+        'ssl_crlfile': None,
     }
 
     def __init__(self, **configs):
@@ -111,6 +112,11 @@ class KafkaClient(object):
                 establish the certificate's authenticity. default: none.
             ssl_keyfile (str): optional filename containing the client private key.
                 default: none.
+            ssl_crlfile (str): optional filename containing the CRL to check for
+                certificate expiration. By default, no CRL check is done. When
+                providing a file, only the leaf certificate will be checked against
+                this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+.
+                default: none.
         """
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
diff --git a/kafka/conn.py b/kafka/conn.py
index db56dda..cf5dce3 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -70,6 +70,7 @@ class BrokerConnection(object):
         'ssl_cafile': None,
         'ssl_certfile': None,
         'ssl_keyfile': None,
+        'ssl_crlfile': None,
         'api_version': (0, 8, 2),  # default to most restrictive
         'state_change_callback': lambda conn: True,
     }
@@ -228,6 +229,16 @@ class BrokerConnection(object):
                 self._ssl_context.load_cert_chain(
                     certfile=self.config['ssl_certfile'],
                     keyfile=self.config['ssl_keyfile'])
+            if self.config['ssl_crlfile']:
+                if not hasattr(ssl, 'VERIFY_CRL_CHECK_LEAF'):
+                    log.error('%s: No CRL support with this version of Python.'
+                              ' Disconnecting.', self)
+                    self.close()
+                    return
+                log.info('%s: Loading SSL CRL from %s', str(self), self.config['ssl_crlfile'])
+                self._ssl_context.load_verify_locations(self.config['ssl_crlfile'])
+                # pylint: disable=no-member
+                self._ssl_context.verify_flags |= ssl.VERIFY_CRL_CHECK_LEAF
         log.debug('%s: wrapping socket in ssl context', str(self))
         try:
             self._sock = self._ssl_context.wrap_socket(
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index eaaafa4..106e96b 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -138,6 +138,11 @@ class KafkaConsumer(six.Iterator):
             establish the certificate's authenticity. default: none.
         ssl_keyfile (str): optional filename containing the client private key.
             default: none.
+        ssl_crlfile (str): optional filename containing the CRL to check for
+            certificate expiration. By default, no CRL check is done. When
+            providing a file, only the leaf certificate will be checked against
+            this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+.
+            default: none.
         api_version (str): specify which kafka API version to use.
             0.9 enables full group coordination features; 0.8.2 enables
             kafka-storage offset commits; 0.8.1 enables zookeeper-storage
@@ -187,6 +192,7 @@ class KafkaConsumer(six.Iterator):
         'ssl_cafile': None,
         'ssl_certfile': None,
         'ssl_keyfile': None,
+        'ssl_crlfile': None,
         'api_version': 'auto',
         'connections_max_idle_ms': 9 * 60 * 1000, # not implemented yet
         'metric_reporters': [],
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index 6d2c816..7e8f625 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -207,6 +207,11 @@ class KafkaProducer(object):
             establish the certificate's authenticity. default: none.
         ssl_keyfile (str): optional filename containing the client private key.
             default: none.
+        ssl_crlfile (str): optional filename containing the CRL to check for
+            certificate expiration. By default, no CRL check is done. When
+            providing a file, only the leaf certificate will be checked against
+            this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+.
+            default: none.
         api_version (str): specify which kafka API version to use.
             If set to 'auto', will attempt to infer the broker version by
             probing various APIs. Default: auto
@@ -243,6 +248,7 @@ class KafkaProducer(object):
         'ssl_cafile': None,
         'ssl_certfile': None,
         'ssl_keyfile': None,
+        'ssl_crlfile': None,
         'api_version': 'auto',
     }