Add HTTP Auth and SSL to the ES output plugin

The typical use case is the placement of a transparent proxy in front of
ES. This PR enables basic Auth and HTTPS to be configured independently.

This patch makes logstash use only elasticsearch-ruby gem when writing
to elasticsearch. Also leverages the manticore http transport that ships
with es-ruby for the basic http auth and ssl features.

It's possible to configure the CA certificate using either a .pem/.cer
file ("cacert" option), or a .jks truststore ("truststore" option).

Author: jsvd

lib/logstash/outputs/elasticsearch.rb

@@ -5,6 +5,7 @@
 require "logstash/json"
 require "stud/buffer"
 require "socket" # for Socket.gethostname
+require "uri" # for escaping user input
 require 'logstash-output-elasticsearch_jars.rb'
 
 # This output lets you store logs in Elasticsearch and is the most recommended
@@ -185,18 +186,28 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   config :plugins, :validate => :array
 
 
+  # Username and password (HTTP only)
+  config :user, :validate => :string
+  config :password, :validate => :password
+
+  # SSL Configurations (HTTP only)
+  #
+  # Enable SSL
+  config :ssl, :validate => :boolean, :default => false
+
+  # The .cer or .pem file to validate the server's certificate
+  config :cacert, :validate => :path
+
+  # The JKS truststore to validate the server's certificate
+  # Use either :truststore or :cacert
+  config :truststore, :validate => :path
+
+  # Set the truststore password
+  config :truststore_password, :validate => :password
+
   public
   def register
     client_settings = {}
-    client_settings["cluster.name"] = @cluster if @cluster
-    client_settings["network.host"] = @bind_host if @bind_host
-    client_settings["transport.tcp.port"] = @bind_port if @bind_port
-
-    if @node_name
-      client_settings["node.name"] = @node_name
-    else
-      client_settings["node.name"] = "logstash-#{Socket.gethostname}-#{$$}-#{object_id}"
-    end
 
     if @protocol.nil?
       @protocol = LogStash::Environment.jruby? ? "node" : "http"
@@ -217,6 +228,15 @@ def register
 
       # setup log4j properties for Elasticsearch
  #     LogStash::Logger.setup_log4j(@logger)
+       client_settings["cluster.name"] = @cluster if @cluster
+       client_settings["network.host"] = @bind_host if @bind_host
+       client_settings["transport.tcp.port"] = @bind_port if @bind_port
+ 
+       if @node_name
+         client_settings["node.name"] = @node_name
+       else
+         client_settings["node.name"] = "logstash-#{Socket.gethostname}-#{$$}-#{object_id}"
+       end
     end
 
     require "logstash/outputs/elasticsearch/protocol"
@@ -233,12 +253,45 @@ def register
       @host = ["localhost"]
     end
 
+    if @ssl
+      if @protocol == "http"
+        @protocol = "https"
+        if @cacert && @truststore
+          raise(LogStash::ConfigurationError, "Use either \"cacert\" or \"truststore\" when configuring the CA certificate") if @truststore
+        end
+        ssl_options = {}
+        if @cacert then
+          @truststore, ssl_options[:truststore_password] = generate_jks @cacert
+        elsif @truststore
+          ssl_options[:truststore_password] = @truststore_password.value if @truststore_password
+        end
+        ssl_options[:truststore] = @truststore
+        client_settings[:ssl] = ssl_options
+      else
+        raise(LogStash::ConfigurationError, "SSL is not supported for '#{@protocol}'. Change the protocol to 'http' if you need SSL.")
+      end
+    end
+
+    common_options = {
+      :protocol => @protocol,
+      :client_settings => client_settings
+    }
+
+    if @user && @password
+      if @protocol =~ /http/
+        common_options[:user] = ::URI.escape(@user, "@:")
+        common_options[:password] = ::URI.escape(@password.value, "@:")
+      else
+        raise(LogStash::ConfigurationError, "User and password parameters are not supported for '#{@protocol}'. Change the protocol to 'http' if you need them.")
+      end
+    end
+
     client_class = case @protocol
       when "transport"
         LogStash::Outputs::Elasticsearch::Protocols::TransportClient
       when "node"
         LogStash::Outputs::Elasticsearch::Protocols::NodeClient
-      when "http"
+      when /http/
         LogStash::Outputs::Elasticsearch::Protocols::HTTPClient
     end
 
@@ -261,17 +314,15 @@ def register
       options = {
           :host => @host,
           :port => @port,
-          :client_settings => client_settings
-      }
+      }.merge(common_options)
       @client << client_class.new(options)
     else # if @protocol in ["transport","http"]
       @host.each do |host|
           (_host,_port) = host.split ":"
           options = {
             :host => _host,
             :port => _port || @port,
-            :client_settings => client_settings
-          }
+          }.merge(common_options)
           @logger.info "Create client to elasticsearch server on #{_host}:#{_port}"
           @client << client_class.new(options)
       end # @host.each
@@ -338,6 +389,29 @@ def start_local_elasticsearch
     @embedded_elasticsearch.start
   end # def start_local_elasticsearch
 
+  private
+  def generate_jks cert_path
+
+    require 'securerandom'
+    require 'tempfile'
+    require 'java'
+    import java.io.FileInputStream
+    import java.io.FileOutputStream
+    import java.security.KeyStore
+    import java.security.cert.CertificateFactory
+
+    jks = java.io.File.createTempFile("cert", ".jks")
+
+    ks = KeyStore.getInstance "JKS"
+    ks.load nil, nil
+    cf = CertificateFactory.getInstance "X.509"
+    cert = cf.generateCertificate FileInputStream.new(cert_path)
+    ks.setCertificateEntry "cacert", cert
+    pwd = SecureRandom.urlsafe_base64(9)
+    ks.store FileOutputStream.new(jks), pwd.to_java.toCharArray
+    [jks.path, pwd]
+  end
+
   public
   def receive(event)
     return unless output?(event)
@@ -376,6 +450,9 @@ def flush(actions, teardown=false)
   end # def flush
 
   def teardown
+    if @cacert # remove temporary jks store created from the cacert
+      File.delete(@truststore)
+    end
     buffer_flush(:final => true)
   end
 

lib/logstash/outputs/elasticsearch/protocol.rb

@@ -51,82 +51,41 @@ class HTTPClient < Base
       }
 
       def initialize(options={})
-        require "ftw"
         super
         require "elasticsearch" # gem 'elasticsearch-ruby'
+        # manticore http transport
+        require "elasticsearch/transport/transport/http/manticore"
         @options = DEFAULT_OPTIONS.merge(options)
         @client = client
       end
 
       def build_client(options)
-        client = Elasticsearch::Client.new(
-          :host => [options[:host], options[:port]].join(":")
-        )
-
-        # Use FTW to do indexing requests, for now, until we
-        # can identify and resolve performance problems of elasticsearch-ruby
-        @bulk_url = "http://#{options[:host]}:#{options[:port]}/_bulk"
-        @agent = FTW::Agent.new
-
-        return client
-      end
-
-      if ENV["BULK"] == "esruby"
-        def bulk(actions)
-          bulk_esruby(actions)
-        end
-      else
-        def bulk(actions)
-          bulk_ftw(actions)
+        uri = "#{options[:protocol]}://#{options[:host]}:#{options[:port]}"
+
+        client_options = {
+          :host => [uri],
+          :transport_options => options[:client_settings]
+        }
+        client_options[:transport_class] = ::Elasticsearch::Transport::Transport::HTTP::Manticore
+        client_options[:ssl] = client_options[:transport_options].delete(:ssl)
+
+        if options[:user] && options[:password] then
+          token = Base64.strict_encode64(options[:user] + ":" + options[:password])
+          client_options[:headers] = { "Authorization" => "Basic #{token}" }
         end
+
+        Elasticsearch::Client.new client_options
       end
 
-      def bulk_esruby(actions)
+      def bulk(actions)
         @client.bulk(:body => actions.collect do |action, args, source|
           if source
             next [ { action => args }, source ]
           else
             next { action => args }
           end
         end.flatten)
-      end # def bulk_esruby
-
-      # Avoid creating a new string for newline every time
-      NEWLINE = "\n".freeze
-      def bulk_ftw(actions)
-        body = actions.collect do |action, args, source|
-          header = { action => args }
-          if source
-            next [ LogStash::Json.dump(header), NEWLINE, LogStash::Json.dump(source), NEWLINE ]
-          else
-            next [ LogStash::Json.dump(header), NEWLINE ]
-          end
-        end.flatten.join("")
-        begin
-          response = @agent.post!(@bulk_url, :body => body)
-        rescue EOFError
-          @logger.warn("EOF while writing request or reading response header from elasticsearch", :host => @host, :port => @port)
-          raise
-        end
-
-        # Consume the body for error checking
-        # This will also free up the connection for reuse.
-        response_body = ""
-        begin
-          response.read_body { |chunk| response_body += chunk }
-        rescue EOFError
-          @logger.warn("EOF while reading response body from elasticsearch",
-                       :url => @bulk_url)
-          raise
-        end
-
-        if response.status != 200
-          @logger.error("Error writing (bulk) to elasticsearch",
-                        :response => response, :response_body => response_body,
-                        :request_body => body)
-          raise "Non-OK response code from Elasticsearch: #{response.status}"
-        end
-      end # def bulk_ftw
+      end # def bulk
 
       def template_exists?(name)
         @client.indices.get_template(:name => name)
@@ -292,4 +251,3 @@ class Index; end
     class Delete; end
   end
 end
-

logstash-output-elasticsearch.gemspec

@@ -23,12 +23,15 @@ Gem::Specification.new do |s|
   s.requirements << "jar 'org.elasticsearch:elasticsearch', '1.3.1'"
 
   # Gem dependencies
-  s.add_runtime_dependency 'elasticsearch'
+  s.add_runtime_dependency 'elasticsearch', ['~> 1.0.6']
   s.add_runtime_dependency 'stud'
   s.add_runtime_dependency 'cabin', ['>=0.6.0']
   s.add_runtime_dependency 'ftw', ['~> 0.0.39']
   s.add_runtime_dependency 'logstash', '>= 1.4.0', '< 2.0.0'
   s.add_runtime_dependency 'jar-dependencies', ['~> 0.0.7']
 
+  if RUBY_PLATFORM == 'java'
+    gem.add_runtime_dependency "manticore"
+  end
 end
 

spec/outputs/elasticsearch.rb

@@ -386,4 +386,105 @@
       end
     end
   end
+
+  describe "Authentication option" do
+    ["node", "transport"].each do |protocol|
+      context "with protocol => #{protocol}" do
+        subject do
+          require "logstash/outputs/elasticsearch"
+          settings = {
+            "protocol" => protocol,
+            "node_name" => "logstash",
+            "cluster" => "elasticsearch",
+            "host" => "node01",
+            "user" => "test",
+            "password" => "test"
+          }
+          next LogStash::Outputs::ElasticSearch.new(settings)
+        end
+
+        it "should fail in register" do
+          expect {subject.register}.to raise_error
+        end
+      end
+    end
+  end
+
+  describe "SSL option" do
+    ["node", "transport"].each do |protocol|
+      context "with protocol => #{protocol}" do
+        subject do
+          require "logstash/outputs/elasticsearch"
+          settings = {
+            "protocol" => protocol,
+            "node_name" => "logstash",
+            "cluster" => "elasticsearch",
+            "host" => "node01",
+            "ssl" => true
+          }
+          next LogStash::Outputs::ElasticSearch.new(settings)
+        end
+
+        it "should fail in register" do
+          expect {subject.register}.to raise_error
+        end
+      end
+    end
+  end
+
+  describe "send messages to ElasticSearch using HTTPS", :elasticsearch_secure => true do
+    subject do
+      require "logstash/outputs/elasticsearch"
+      settings = {
+        "protocol" => "http",
+        "node_name" => "logstash",
+        "cluster" => "elasticsearch",
+        "host" => "node01",
+        "user" => "user",
+        "password" => "changeme",
+        "ssl" => true,
+        "cacert" => "/tmp/ca/certs/cacert.pem",
+        # or
+        #"truststore" => "/tmp/ca/truststore.jks",
+        #"truststore_password" => "testeteste"
+      }
+      next LogStash::Outputs::ElasticSearch.new(settings)
+    end
+
+    before :each do
+      subject.register
+    end
+
+    it "sends events to ES" do
+      expect {
+        subject.receive(LogStash::Event.new("message" => "sample message here"))
+        subject.buffer_flush(:final => true)
+      }.to_not raise_error
+    end
+  end
+
+  describe "connect using HTTP Authentication", :elasticsearch_secure => true do
+    subject do
+      require "logstash/outputs/elasticsearch"
+      settings = {
+        "protocol" => "http",
+        "cluster" => "elasticsearch",
+        "host" => "node01",
+        "user" => "user",
+        "password" => "changeme",
+      }
+      next LogStash::Outputs::ElasticSearch.new(settings)
+    end
+
+    before :each do
+      subject.register
+    end
+
+    it "sends events to ES" do
+      expect {
+        subject.receive(LogStash::Event.new("message" => "sample message here"))
+        subject.buffer_flush(:final => true)
+      }.to_not raise_error
+    end
+  end
 end