Enable SCRAM-SHA-256 and SCRAM-SHA-512 for sasl

.travis.yml

@@ -25,7 +25,7 @@ addons:
 cache:
  directories:
  - $HOME/.cache/pip
- - servers/
+ - servers/dist
 
 before_install:
  - source travis_java_install.sh

kafka/admin/client.py

@@ -130,11 +130,11 @@ class KafkaAdminClient(object):
  metric_group_prefix (str): Prefix for metric names. Default: ''
  sasl_mechanism (str): Authentication mechanism when security_protocol
  is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are:
- PLAIN, GSSAPI, OAUTHBEARER.
- sasl_plain_username (str): username for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
- sasl_plain_password (str): password for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
+ PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512.
+ sasl_plain_username (str): username for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
+ sasl_plain_password (str): password for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
  sasl_kerberos_service_name (str): Service name to include in GSSAPI
  sasl mechanism handshake. Default: 'kafka'
  sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI

kafka/client_async.py

@@ -143,11 +143,11 @@ class KafkaClient(object):
  metric_group_prefix (str): Prefix for metric names. Default: ''
  sasl_mechanism (str): Authentication mechanism when security_protocol
  is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are:
- PLAIN, GSSAPI, OAUTHBEARER.
- sasl_plain_username (str): username for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
- sasl_plain_password (str): password for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
+ PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512.
+ sasl_plain_username (str): username for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
+ sasl_plain_password (str): password for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
  sasl_kerberos_service_name (str): Service name to include in GSSAPI
  sasl mechanism handshake. Default: 'kafka'
  sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
@@ -767,10 +767,7 @@ def least_loaded_node(self):
  inflight = curr_inflight
  found = node_id
 
- if found is not None:
- return found
-
- return None
+ return found
 
  def set_topics(self, topics):
  """Set specific topics to track for metadata.

kafka/conn.py

@@ -1,12 +1,16 @@
 from __future__ import absolute_import, division
 
-import collections
+import base64
 import copy
 import errno
+import hashlib
+import hmac
 import io
 import logging
 from random import shuffle, uniform
 
+from uuid import uuid4
+
 # selectors in stdlib as of py3.4
 try:
  import selectors # pylint: disable=import-error
@@ -16,7 +20,6 @@
 
 import socket
 import struct
-import sys
 import threading
 import time
 
@@ -39,6 +42,12 @@
  TimeoutError = socket.error
  BlockingIOError = Exception
 
+ def xor_bytes(left, right):
+ return bytearray(ord(lb) ^ ord(rb) for lb, rb in zip(left, right))
+else:
+ def xor_bytes(left, right):
+ return bytes(lb ^ rb for lb, rb in zip(left, right))
+
 log = logging.getLogger(__name__)
 
 DEFAULT_KAFKA_PORT = 9092
@@ -98,6 +107,69 @@ class ConnectionStates(object):
  AUTHENTICATING = '<authenticating>'
 
 
+class ScramClient:
+ MECHANISMS = {
+ 'SCRAM-SHA-256': hashlib.sha256,
+ 'SCRAM-SHA-512': hashlib.sha512
+ }
+
+ def __init__(self, user, password, mechanism):
+ self.nonce = str(uuid4()).replace('-', '')
+ self.auth_message = ''
+ self.salted_password = None
+ self.user = user
+ self.password = password.encode()
+ self.hashfunc = self.MECHANISMS[mechanism]
+ self.hashname = ''.join(mechanism.lower().split('-')[1:3])
+ self.stored_key = None
+ self.client_key = None
+ self.client_signature = None
+ self.client_proof = None
+ self.server_key = None
+ self.server_signature = None
+
+ def first_message(self):
+ client_first_bare = 'n={},r={}'.format(self.user, self.nonce)
+ self.auth_message += client_first_bare
+ return 'n,,' + client_first_bare
+
+ def process_server_first_message(self, server_first_message):
+ self.auth_message += ',' + server_first_message
+ params = dict(pair.split('=', 1) for pair in server_first_message.split(','))
+ server_nonce = params['r']
+ if not server_nonce.startswith(self.nonce):
+ raise ValueError("Server nonce, did not start with client nonce!")
+ self.nonce = server_nonce
+ self.auth_message += ',c=biws,r=' + self.nonce
+
+ salt = base64.b64decode(params['s'].encode())
+ iterations = int(params['i'])
+ self.create_salted_password(salt, iterations)
+
+ self.client_key = self.hmac(self.salted_password, b'Client Key')
+ self.stored_key = self.hashfunc(self.client_key).digest()
+ self.client_signature = self.hmac(self.stored_key, self.auth_message.encode())
+ self.client_proof = xor_bytes(self.client_key, self.client_signature)
+ self.server_key = self.hmac(self.salted_password, b'Server Key')
+ self.server_signature = self.hmac(self.server_key, self.auth_message.encode())
+
+ def hmac(self, key, msg):
+ return hmac.new(key, msg, digestmod=self.hashfunc).digest()
+
+ def create_salted_password(self, salt, iterations):
+ self.salted_password = hashlib.pbkdf2_hmac(
+ self.hashname, self.password, salt, iterations
+ )
+
+ def final_message(self):
+ client_final_no_proof = 'c=biws,r=' + self.nonce
+ return 'c=biws,r={},p={}'.format(self.nonce, base64.b64encode(self.client_proof).decode())
+
+ def process_server_final_message(self, server_final_message):
+ params = dict(pair.split('=', 1) for pair in server_final_message.split(','))
+ if self.server_signature != base64.b64decode(params['v'].encode()):
+ raise ValueError("Server sent wrong signature!")
+
 class BrokerConnection(object):
  """Initialize a Kafka broker connection
 
@@ -177,11 +249,11 @@ class BrokerConnection(object):
  metric_group_prefix (str): Prefix for metric names. Default: ''
  sasl_mechanism (str): Authentication mechanism when security_protocol
  is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are:
- PLAIN, GSSAPI, OAUTHBEARER.
- sasl_plain_username (str): username for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
- sasl_plain_password (str): password for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
+ PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512.
+ sasl_plain_username (str): username for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
+ sasl_plain_password (str): password for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
  sasl_kerberos_service_name (str): Service name to include in GSSAPI
  sasl mechanism handshake. Default: 'kafka'
  sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
@@ -224,7 +296,7 @@ class BrokerConnection(object):
  'sasl_oauth_token_provider': None
  }
  SECURITY_PROTOCOLS = ('PLAINTEXT', 'SSL', 'SASL_PLAINTEXT', 'SASL_SSL')
- SASL_MECHANISMS = ('PLAIN', 'GSSAPI', 'OAUTHBEARER')
+ SASL_MECHANISMS = ('PLAIN', 'GSSAPI', 'OAUTHBEARER', "SCRAM-SHA-256", "SCRAM-SHA-512")
 
  def __init__(self, host, port, afi, **configs):
  self.host = host
@@ -259,9 +331,13 @@ def __init__(self, host, port, afi, **configs):
  if self.config['security_protocol'] in ('SASL_PLAINTEXT', 'SASL_SSL'):
  assert self.config['sasl_mechanism'] in self.SASL_MECHANISMS, (
  'sasl_mechanism must be in ' + ', '.join(self.SASL_MECHANISMS))
- if self.config['sasl_mechanism'] == 'PLAIN':
- assert self.config['sasl_plain_username'] is not None, 'sasl_plain_username required for PLAIN sasl'
- assert self.config['sasl_plain_password'] is not None, 'sasl_plain_password required for PLAIN sasl'
+ if self.config['sasl_mechanism'] in ('PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512'):
+ assert self.config['sasl_plain_username'] is not None, (
+ 'sasl_plain_username required for PLAIN or SCRAM sasl'
+ )
+ assert self.config['sasl_plain_password'] is not None, (
+ 'sasl_plain_password required for PLAIN or SCRAM sasl'
+ )
  if self.config['sasl_mechanism'] == 'GSSAPI':
  assert gssapi is not None, 'GSSAPI lib not available'
  assert self.config['sasl_kerberos_service_name'] is not None, 'sasl_kerberos_service_name required for GSSAPI sasl'
@@ -552,6 +628,8 @@ def _handle_sasl_handshake_response(self, future, response):
  return self._try_authenticate_gssapi(future)
  elif self.config['sasl_mechanism'] == 'OAUTHBEARER':
  return self._try_authenticate_oauth(future)
+ elif self.config['sasl_mechanism'].startswith("SCRAM-SHA-"):
+ return self._try_authenticate_scram(future)
  else:
  return future.failure(
  Errors.UnsupportedSaslMechanismError(
@@ -652,6 +730,53 @@ def _try_authenticate_plain(self, future):
  log.info('%s: Authenticated as %s via PLAIN', self, self.config['sasl_plain_username'])
  return future.success(True)
 
+ def _try_authenticate_scram(self, future):
+ if self.config['security_protocol'] == 'SASL_PLAINTEXT':
+ log.warning('%s: Exchanging credentials in the clear', self)
+
+ scram_client = ScramClient(
+ self.config['sasl_plain_username'], self.config['sasl_plain_password'], self.config['sasl_mechanism']
+ )
+
+ err = None
+ close = False
+ with self._lock:
+ if not self._can_send_recv():
+ err = Errors.NodeNotReadyError(str(self))
+ close = False
+ else:
+ try:
+ client_first = scram_client.first_message().encode()
+ size = Int32.encode(len(client_first))
+ self._send_bytes_blocking(size + client_first)
+
+ (data_len,) = struct.unpack('>i', self._recv_bytes_blocking(4))
+ server_first = self._recv_bytes_blocking(data_len).decode()
+ scram_client.process_server_first_message(server_first)
+
+ client_final = scram_client.final_message().encode()
+ size = Int32.encode(len(client_final))
+ self._send_bytes_blocking(size + client_final)
+
+ (data_len,) = struct.unpack('>i', self._recv_bytes_blocking(4))
+ server_final = self._recv_bytes_blocking(data_len).decode()
+ scram_client.process_server_final_message(server_final)
+
+ except (ConnectionError, TimeoutError) as e:
+ log.exception("%s: Error receiving reply from server", self)
+ err = Errors.KafkaConnectionError("%s: %s" % (self, e))
+ close = True
+
+ if err is not None:
+ if close:
+ self.close(error=err)
+ return future.failure(err)
+
+ log.info(
+ '%s: Authenticated as %s via %s', self, self.config['sasl_plain_username'], self.config['sasl_mechanism']
+ )
+ return future.success(True)
+
  def _try_authenticate_gssapi(self, future):
  kerberos_damin_name = self.config['sasl_kerberos_domain_name'] or self.host
  auth_id = self.config['sasl_kerberos_service_name'] + '@' + kerberos_damin_name

kafka/consumer/group.py

@@ -231,11 +231,11 @@ class KafkaConsumer(six.Iterator):
  subscribing to it. Requires 0.10+ Default: True
  sasl_mechanism (str): Authentication mechanism when security_protocol
  is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are:
- PLAIN, GSSAPI, OAUTHBEARER.
- sasl_plain_username (str): Username for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
- sasl_plain_password (str): Password for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
+ PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512.
+ sasl_plain_username (str): username for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
+ sasl_plain_password (str): password for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
  sasl_kerberos_service_name (str): Service name to include in GSSAPI
  sasl mechanism handshake. Default: 'kafka'
  sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI

kafka/producer/kafka.py

@@ -268,11 +268,11 @@ class KafkaProducer(object):
  Default: selectors.DefaultSelector
  sasl_mechanism (str): Authentication mechanism when security_protocol
  is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are:
- PLAIN, GSSAPI, OAUTHBEARER.
- sasl_plain_username (str): username for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
- sasl_plain_password (str): password for sasl PLAIN authentication.
- Required if sasl_mechanism is PLAIN.
+ PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512.
+ sasl_plain_username (str): username for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
+ sasl_plain_password (str): password for sasl PLAIN and SCRAM authentication.
+ Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms.
  sasl_kerberos_service_name (str): Service name to include in GSSAPI
  sasl mechanism handshake. Default: 'kafka'
  sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI

requirements-dev.txt

@@ -8,6 +8,7 @@ lz4==2.1.2
 xxhash==1.3.0
 python-snappy==0.5.3
 tox==3.5.3
+mock==3.0.5
 pylint==1.9.3
 pytest-pylint==0.12.3
 pytest-mock==1.10.0

servers/0.10.0.0/resources/kafka.properties

@@ -24,6 +24,8 @@ broker.id={broker_id}
 listeners={transport}://{host}:{port}
 security.inter.broker.protocol={transport}
 
+{sasl_config}
+
 ssl.keystore.location={ssl_dir}/kafka.server.keystore.jks
 ssl.keystore.password=foobar
 ssl.key.password=foobar
@@ -121,7 +123,7 @@ log.cleaner.enable=false
 # tune down offset topics to reduce setup time in tests
 offsets.commit.timeout.ms=500
 offsets.topic.num.partitions=2
-offsets.topic.replication.factor=2
+offsets.topic.replication.factor=1
 
 # Allow shorter session timeouts for tests
 group.min.session.timeout.ms=1000

servers/0.10.0.0/resources/kafka_server_jaas.conf

@@ -0,0 +1,4 @@
+KafkaServer {{
+ {jaas_config}
+}};
+Client {{}};

servers/0.10.0.1/resources/kafka.properties

@@ -24,6 +24,8 @@ broker.id={broker_id}
 listeners={transport}://{host}:{port}
 security.inter.broker.protocol={transport}
 
+{sasl_config}
+
 ssl.keystore.location={ssl_dir}/kafka.server.keystore.jks
 ssl.keystore.password=foobar
 ssl.key.password=foobar
@@ -121,7 +123,7 @@ log.cleaner.enable=false
 # tune down offset topics to reduce setup time in tests
 offsets.commit.timeout.ms=500
 offsets.topic.num.partitions=2
-offsets.topic.replication.factor=2
+offsets.topic.replication.factor=1
 
 # Allow shorter session timeouts for tests
 group.min.session.timeout.ms=1000

servers/0.10.0.1/resources/kafka_server_jaas.conf

@@ -0,0 +1,4 @@
+KafkaServer {{
+ {jaas_config}
+}};
+Client {{}};

servers/0.10.1.1/resources/kafka.properties

@@ -24,6 +24,8 @@ broker.id={broker_id}
 listeners={transport}://{host}:{port}
 security.inter.broker.protocol={transport}
 
+{sasl_config}
+
 ssl.keystore.location={ssl_dir}/kafka.server.keystore.jks
 ssl.keystore.password=foobar
 ssl.key.password=foobar
@@ -121,7 +123,7 @@ log.cleaner.enable=false
 # tune down offset topics to reduce setup time in tests
 offsets.commit.timeout.ms=500
 offsets.topic.num.partitions=2
-offsets.topic.replication.factor=2
+offsets.topic.replication.factor=1
 
 # Allow shorter session timeouts for tests
 group.min.session.timeout.ms=1000

servers/0.10.1.1/resources/kafka_server_jaas.conf

@@ -0,0 +1,4 @@
+KafkaServer {{
+ {jaas_config}
+}};
+Client {{}};