feat(kafka): RHICOMPL-1893 Kafka SASL authentication

app/producers/application_producer.rb

@@ -27,15 +27,27 @@ def logger
     end
 
     def kafka_ca_cert
-      return unless Settings.kafka.security_protocol == 'ssl'
+      return unless %w[ssl sasl_ssl].include?(Settings.kafka.security_protocol)
 
       File.read(Settings.kafka.ssl_ca_location)
     end
 
+    def sasl_config
+      return unless Settings.kafka.security_protocol == 'sasl_ssl'
+
+      {
+        sasl_scram_username: Settings.kafka.sasl_username,
+        sasl_scram_password: Settings.kafka.sasl_password,
+        sasl_scram_mechanism: 'sha512'
+      }
+    end
+
     def kafka_config
       {}.tap do |config|
         config[:client_id] = self::CLIENT_ID
         config[:ssl_ca_cert] = kafka_ca_cert if kafka_ca_cert
+
+        config.merge!(sasl_config) if sasl_config
       end
     end
 

config/racecar.rb

@@ -4,4 +4,10 @@
 
   config.security_protocol = Settings.kafka.security_protocol
   config.ssl_ca_location = Settings.kafka.ssl_ca_location
+
+  if Settings.kafka.security_protocol == 'sasl_ssl'
+    config.sasl_username = Settings.kafka.sasl_username
+    config.sasl_password = Settings.kafka.sasl_password
+    config.sasl_mechanism = 'SCRAM-SHA-512'
+  end
 end

config/settings.yml

@@ -5,6 +5,8 @@ kafka:
   brokers: ''
   security_protocol: plaintext
   ssl_ca_location:
+  sasl_username:
+  sasl_password:
 kafka_consumer_topics:
   inventory_events: 'platform.inventory.events'
 kafka_producer_topics:

test/producers/application_producer_test.rb

@@ -20,6 +20,25 @@ class MockProducer < ApplicationProducer; end
     assert_equal config, MockProducer.send(:kafka_config)
   end
 
+  test 'handles SASL SSL settings' do
+    Settings.kafka.security_protocol = 'sasl_ssl'
+    Settings.kafka.ssl_ca_location = 'test/fixtures/files/test_ca.crt'
+    Settings.kafka.sasl_username = 'user'
+    Settings.kafka.sasl_password = 'youwish'
+
+    class MockProducer < ApplicationProducer; end
+
+    config = {
+      client_id: ApplicationProducer::CLIENT_ID,
+      ssl_ca_cert: "very secure\n",
+      sasl_scram_username: 'user',
+      sasl_scram_password: 'youwish',
+      sasl_scram_mechanism: 'sha512'
+    }
+
+    assert_equal config, MockProducer.send(:kafka_config)
+  end
+
   test 'handles plaintext settings' do
     Settings.kafka.security_protocol = 'plaintext'