validate sasl_mechanism against SaslHandShakeResponse enabled_mechanismsauth_recv

1 files changed, 6 insertions, 1 deletions

diff --git a/kafka/conn.py b/kafka/conn.py
index 2f560d0..7ca2652 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -460,7 +460,12 @@ class BrokerConnection(object):
             self.close(error=error)
             return future.failure(error_type(self))
 
-        if self.config['sasl_mechanism'] == 'PLAIN':
+        if self.config['sasl_mechanism'] not in response.enabled_mechanisms:
+            return future.failure(
+                Errors.UnsupportedSaslMechanismError(
+                    'Kafka broker does not support %s sasl mechanism. Enabled mechanisms are: %s'
+                    % (self.config['sasl_mechanism'], response.enabled_mechanisms)))
+        elif self.config['sasl_mechanism'] == 'PLAIN':
             return self._try_authenticate_plain(future)
         elif self.config['sasl_mechanism'] == 'GSSAPI':
             return self._try_authenticate_gssapi(future)