Fix GSSAPI authentication

1 files changed, 32 insertions, 42 deletions

diff --git a/kafka/conn.py b/kafka/conn.py
index dbe212a..067dc0f 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -514,52 +514,42 @@ class BrokerConnection(object):
         ctx_CanonName = ctx_Name.canonicalize(gssapi.MechType.kerberos)
         log.debug('%s: canonical Servicename: %s', self, ctx_CanonName)
         ctx_Context = gssapi.SecurityContext(name=ctx_CanonName, usage='initiate')
-        # Exchange tokens until authentication either succeeds or fails:
+        log.debug("%s: initiator name: %s", self, ctx_Context.initiator_name)
+
+        # Exchange tokens until authentication either succeeds or fails
         received_token = None
-        try:
-            while not ctx_Context.complete:
-                # calculate the output token
-                try:
-                    output_token = ctx_Context.step(received_token)
-                except GSSError as e:
-                    log.exception("%s: Error invalid token received from server",  self)
-                    error = Errors.ConnectionError("%s: %s" % (self, e))
-
-                if not output_token:
-                    if ctx_Context.complete:
-                        log.debug("%s: Security Context complete ", self)
-                    log.debug("%s: Successful GSSAPI handshake for %s", self, ctx_Context.initiator_name)
-                    break
-                try:
-                    self._sock.setblocking(True)
-                    # Send output token
-                    msg = output_token
-                    size = Int32.encode(len(msg))
-                    self._sock.sendall(size + msg)
-
-                    # The server will send a token back. Processing of this token either
-                    # establishes a security context, or it needs further token exchange.
-                    # The gssapi will be able to identify the needed next step.
-                    # The connection is closed on failure.
-                    response = self._sock.recv(2000)
-                    self._sock.setblocking(False)
-
-                except (AssertionError, ConnectionError) as e:
-                    log.exception("%s: Error receiving reply from server",  self)
-                    error = Errors.ConnectionError("%s: %s" % (self, e))
-                    future.failure(error)
-                    self.close(error=error)
+        while not ctx_Context.complete:
+            # calculate an output token from kafka token (or None if first iteration)
+            # exceptions raised here are uncaught and will be sent to the user
+            output_token = ctx_Context.step(received_token)
 
-                # pass the received token back to gssapi, strip the first 4 bytes
-                received_token = response[4:]
+            # pass output token to kafka
+            try:
+                self._sock.setblocking(True)
+                msg = output_token
+                size = Int32.encode(len(msg))
+                self._sock.sendall(size + msg)
+                # The server will send a token back. Processing of this token either
+                # establishes a security context, or it needs further token exchange.
+                # The gssapi will be able to identify the needed next step.
+                # The connection is closed on failure.
+                response = self._sock.recv(2000)
+                self._sock.setblocking(False)
 
-        except Exception as e:
-            log.exception("%s: GSSAPI handshake error",  self)
-            error = Errors.ConnectionError("%s: %s" % (self, e))
-            future.failure(error)
-            self.close(error=error)
+            except (AssertionError, ConnectionError) as e:
+                log.exception("%s: Error receiving reply from server",  self)
+                error = Errors.ConnectionError("%s: %s" % (self, e))
+                self.close(error=error)
+                return future.failure(error)
+
+            # pass the received token back to gssapi, strip the first 4 bytes
+            # dpkp note: what are the first four bytes here?
+            # it seems likely that this is the encoded message size
+            # which we should receive and parse first, then use to parse
+            # the remainder of the token
+            received_token = response[4:]
 
-        log.info('%s: Authenticated as %s', self, gssname)
+        log.info('%s: Authenticated as %s via GSSAPI', self, gssname)
         return future.success(True)
 
     def blacked_out(self):