Sha256: 24b70dccc9d10f9025b2b7623ddbe7d2380631b300e7accc561616ba304a6236
Contents?: true
Size: 1.13 KB
Versions: 12
Compression:
Stored size: 1.13 KB
Contents
module Kafka
class SaslPlainAuthenticator
PLAIN_IDENT = "PLAIN"
def initialize(connection:, logger:, authzid:, username:, password:)
@connection = connection
@logger = logger
@authzid = authzid
@username = username
@password = password
end
def authenticate!
response = @connection.send_request(Kafka::Protocol::SaslHandshakeRequest.new(PLAIN_IDENT))
@encoder = @connection.encoder
@decoder = @connection.decoder
unless response.error_code == 0 && response.enabled_mechanisms.include?(PLAIN_IDENT)
raise Kafka::Error, "#{PLAIN_IDENT} is not supported."
end
# SASL PLAIN
msg = [@authzid.to_s,
@username.to_s,
@password.to_s].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
Version data entries
12 entries across 12 versions & 1 rubygems