Refactor SASL authentication code

Author: dasch

lib/kafka/sasl/gssapi.rb

@@ -0,0 +1,74 @@
+module Kafka
+  module Sasl
+    class Gssapi
+      GSSAPI_IDENT = "GSSAPI"
+      GSSAPI_CONFIDENTIALITY = false
+
+      def initialize(logger:, principal:, keytab:)
+        @logger = logger
+        @principal = principal
+        @keytab = keytab
+      end
+
+      def configured?
+        @principal && !@principal.empty?
+      end
+
+      def ident
+        GSSAPI_IDENT
+      end
+
+      def authenticate!(host, encoder, decoder)
+        load_gssapi
+        initialize_gssapi_context(host)
+
+        @encoder = encoder
+        @decoder = decoder
+
+        # send gssapi token and receive token to verify
+        token_to_verify = send_and_receive_sasl_token
+
+        # verify incoming token
+        unless @gssapi_ctx.init_context(token_to_verify)
+          raise Kafka::Error, "GSSAPI context verification failed."
+        end
+
+        # we can continue, so send OK
+        @encoder.write([0, 2].pack('l>c'))
+
+        # read wrapped message and return it back with principal
+        handshake_messages
+      end
+
+      def handshake_messages
+        msg = @decoder.bytes
+        raise Kafka::Error, "GSSAPI negotiation failed." unless msg
+        # unwrap with integrity only
+        msg_unwrapped = @gssapi_ctx.unwrap_message(msg, GSSAPI_CONFIDENTIALITY)
+        msg_wrapped = @gssapi_ctx.wrap_message(msg_unwrapped + @principal, GSSAPI_CONFIDENTIALITY)
+        @encoder.write_bytes(msg_wrapped)
+      end
+
+      def send_and_receive_sasl_token
+        @encoder.write_bytes(@gssapi_token)
+        @decoder.bytes
+      end
+
+      def load_gssapi
+        begin
+          require "gssapi"
+        rescue LoadError
+          @logger.error "In order to use GSSAPI authentication you need to install the `gssapi` gem."
+          raise
+        end
+      end
+
+      def initialize_gssapi_context(host)
+        @logger.debug "GSSAPI: Initializing context with #{host}, principal #{@principal}"
+
+        @gssapi_ctx = GSSAPI::Simple.new(host, @principal, @keytab)
+        @gssapi_token = @gssapi_ctx.init_context(nil)
+      end
+    end
+  end
+end

lib/kafka/sasl/plain.rb

@@ -0,0 +1,37 @@
+module Kafka
+  module Sasl
+    class Plain
+      PLAIN_IDENT = "PLAIN"
+
+      def initialize(logger:, authzid:, username:, password:)
+        @logger = logger
+        @authzid = authzid
+        @username = username
+        @password = password
+      end
+
+      def ident
+        PLAIN_IDENT
+      end
+
+      def configured?
+        @authzid && @username && @password
+      end
+
+      def authenticate!(host, encoder, decoder)
+        msg = [@authzid, @username, @password].join("\000").force_encoding("utf-8")
+
+        encoder.write_bytes(msg)
+
+        begin
+          msg = decoder.bytes
+          raise Kafka::Error, "SASL PLAIN authentication failed: unknown error" unless msg
+        rescue Errno::ETIMEDOUT, EOFError => e
+          raise Kafka::Error, "SASL PLAIN authentication failed: #{e.message}"
+        end
+
+        @logger.debug "SASL PLAIN authentication successful."
+      end
+    end
+  end
+end

lib/kafka/sasl/scram.rb

@@ -0,0 +1,175 @@
+require 'securerandom'
+require 'base64'
+
+module Kafka
+  module Sasl
+    class Scram
+      MECHANISMS = {
+        "sha256" => "SCRAM-SHA-256",
+        "sha512" => "SCRAM-SHA-512",
+      }.freeze
+
+      def initialize(username:, password:, mechanism: 'sha256', logger:)
+        @username = username
+        @password = password
+        @logger = logger
+
+        if mechanism
+          @mechanism = MECHANISMS.fetch(mechanism) do
+            raise Kafka::SaslScramError, "SCRAM mechanism #{mechanism} is not supported."
+          end
+        end
+      end
+
+      def ident
+        @mechanism
+      end
+
+      def configured?
+        @username && @password && @mechanism
+      end
+
+      def authenticate!(host, encoder, decoder)
+        @logger.debug "Authenticating #{@username} with SASL #{@mechanism}"
+
+        begin
+          msg = first_message
+          @logger.debug "Sending first client SASL SCRAM message: #{msg}"
+          encoder.write_bytes(msg)
+
+          @server_first_message = decoder.bytes
+          @logger.debug "Received first server SASL SCRAM message: #{@server_first_message}"
+
+          msg = final_message
+          @logger.debug "Sending final client SASL SCRAM message: #{msg}"
+          encoder.write_bytes(msg)
+
+          response = parse_response(decoder.bytes)
+          @logger.debug "Received last server SASL SCRAM message: #{response}"
+
+          raise FailedScramAuthentication, response['e'] if response['e']
+          raise FailedScramAuthentication, "Invalid server signature" if response['v'] != server_signature
+        rescue EOFError => e
+          raise FailedScramAuthentication, e.message
+        end
+
+        @logger.debug "SASL SCRAM authentication successful"
+      end
+
+      private
+
+      def first_message
+        "n,,#{first_message_bare}"
+      end
+
+      def first_message_bare
+        "n=#{encoded_username},r=#{nonce}"
+      end
+
+      def final_message_without_proof
+        "c=biws,r=#{rnonce}"
+      end
+
+      def final_message
+        "#{final_message_without_proof},p=#{client_proof}"
+      end
+
+      def server_data
+        parse_response(@server_first_message)
+      end
+
+      def rnonce
+        server_data['r']
+      end
+
+      def salt
+        Base64.strict_decode64(server_data['s'])
+      end
+
+      def iterations
+        server_data['i'].to_i
+      end
+
+      def auth_message
+        [first_message_bare, @server_first_message, final_message_without_proof].join(',')
+      end
+
+      def salted_password
+        hi(@password, salt, iterations)
+      end
+
+      def client_key
+        hmac(salted_password, 'Client Key')
+      end
+
+      def stored_key
+        h(client_key)
+      end
+
+      def server_key
+        hmac(salted_password, 'Server Key')
+      end
+
+      def client_signature
+        hmac(stored_key, auth_message)
+      end
+
+      def server_signature
+        Base64.strict_encode64(hmac(server_key, auth_message))
+      end
+
+      def client_proof
+        Base64.strict_encode64(xor(client_key, client_signature))
+      end
+
+      def h(str)
+        digest.digest(str)
+      end
+
+      def hi(str, salt, iterations)
+        OpenSSL::PKCS5.pbkdf2_hmac(
+          str,
+          salt,
+          iterations,
+          digest.size,
+          digest
+        )
+      end
+
+      def hmac(data, key)
+        OpenSSL::HMAC.digest(digest, data, key)
+      end
+
+      def xor(first, second)
+        first.bytes.zip(second.bytes).map { |(a, b)| (a ^ b).chr }.join('')
+      end
+
+      def parse_response(data)
+        data.split(',').map { |s| s.split('=', 2) }.to_h
+      end
+
+      def encoded_username
+        safe_str(@username.encode(Encoding::UTF_8))
+      end
+
+      def nonce
+        @nonce ||= SecureRandom.urlsafe_base64(32)
+      end
+
+      def digest
+        @digest ||= case @mechanism
+                    when 'SCRAM-SHA-256'
+                      OpenSSL::Digest::SHA256.new
+                    when 'SCRAM-SHA-512'
+                      OpenSSL::Digest::SHA512.new
+                    else
+                      raise ArgumentError, "Unknown SASL mechanism '#{@mechanism}'"
+                    end
+      end
+
+      def safe_str(val)
+        val.gsub('=', '=3D').gsub(',', '=2C')
+      end
+    end
+  end
+end

lib/kafka/sasl_authenticator.rb

@@ -1,80 +1,48 @@
-require 'kafka/sasl_gssapi_authenticator'
-require 'kafka/sasl_plain_authenticator'
-require 'kafka/sasl_scram_authenticator'
+require 'kafka/sasl/plain'
+require 'kafka/sasl/gssapi'
+require 'kafka/sasl/scram'
 
 module Kafka
   class SaslAuthenticator
     def initialize(logger:, sasl_gssapi_principal:, sasl_gssapi_keytab:,
                    sasl_plain_authzid:, sasl_plain_username:, sasl_plain_password:,
                    sasl_scram_username:, sasl_scram_password:, sasl_scram_mechanism:)
       @logger = logger
-      @sasl_gssapi_principal = sasl_gssapi_principal
-      @sasl_gssapi_keytab = sasl_gssapi_keytab
-      @sasl_plain_authzid = sasl_plain_authzid
-      @sasl_plain_username = sasl_plain_username
-      @sasl_plain_password = sasl_plain_password
-      @sasl_scram_username = sasl_scram_username
-      @sasl_scram_password = sasl_scram_password
-      @sasl_scram_mechanism = sasl_scram_mechanism
-    end
-
-    def authenticate!(connection)
-      if authenticate_using_sasl_gssapi?
-        sasl_gssapi_authenticate(connection)
-      elsif authenticate_using_sasl_plain?
-        sasl_plain_authenticate(connection)
-      elsif authenticate_using_sasl_scram?
-        sasl_scram_authenticate(connection)
-      end
-    end
 
-    private
-
-    def sasl_scram_authenticate(connection)
-      auth = SaslScramAuthenticator.new(
-        username: @sasl_scram_username,
-        password: @sasl_scram_password,
+      @plain = Sasl::Plain.new(
+        authzid: sasl_plain_authzid,
+        username: sasl_plain_username,
+        password: sasl_plain_password,
         logger: @logger,
-        mechanism: @sasl_scram_mechanism,
-        connection: connection
       )
 
-      auth.authenticate!
-    end
-
-    def sasl_gssapi_authenticate(connection)
-      auth = SaslGssapiAuthenticator.new(
+      @gssapi = Sasl::Gssapi.new(
+        principal: sasl_gssapi_principal,
+        keytab: sasl_gssapi_keytab,
         logger: @logger,
-        sasl_gssapi_principal: @sasl_gssapi_principal,
-        sasl_gssapi_keytab: @sasl_gssapi_keytab,
-        connection: connection
       )
 
-      auth.authenticate!
-    end
-
-    def sasl_plain_authenticate(connection)
-      auth = SaslPlainAuthenticator.new(
+      @scram = Sasl::Scram.new(
+        username: sasl_scram_username,
+        password: sasl_scram_password,
+        mechanism: sasl_scram_mechanism,
         logger: @logger,
-        authzid: @sasl_plain_authzid,
-        username: @sasl_plain_username,
-        password: @sasl_plain_password,
-        connection: connection
       )
-
-      auth.authenticate!
     end
 
-    def authenticate_using_sasl_scram?
-      @sasl_scram_username && @sasl_scram_password
-    end
+    def authenticate!(connection)
+      mechanism = [@gssapi, @plain, @scram].find(&:configured?)
 
-    def authenticate_using_sasl_gssapi?
-      !@ssl_context && @sasl_gssapi_principal && !@sasl_gssapi_principal.empty?
-    end
+      return if mechanism.nil?
+
+      ident = mechanism.ident
+      response = connection.send_request(Kafka::Protocol::SaslHandshakeRequest.new(ident))
+
+      unless response.error_code == 0 && response.enabled_mechanisms.include?(ident)
+        raise Kafka::Error, "#{ident} is not supported."
+      end
 
-    def authenticate_using_sasl_plain?
-      @sasl_plain_authzid && @sasl_plain_username && @sasl_plain_password
+      mechanism.authenticate!(connection.to_s, connection.encoder, connection.decoder)
     end
   end
 end