add api_key support (elastic#934)

bump version to 10.5.0

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 10.5.0
+  - Added api_key support [#934](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/934)
+
 ## 10.4.1
  - [DOC] Added note about `_type` setting change from `doc` to `_doc` [#884](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/884)
 

docs/index.asciidoc

@@ -232,6 +232,9 @@ Elasticsearch] to take advantage of response compression when using this plugin
 For requests compression, regardless of the Elasticsearch version, users have to enable `http_compression` 
 setting in their Logstash config file.
 
+==== Authentication
+
+Authentication to a secure Elasticsearch cluster is possible using one of the `user`/`password`, `cloud_auth` or `api_key` options.
 
 [id="plugins-{type}s-{plugin}-options"]
 ==== Elasticsearch Output Configuration Options
@@ -242,6 +245,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 |=======================================================================
 |Setting |Input type|Required
 | <<plugins-{type}s-{plugin}-action>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-api_key>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-bulk_path>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-cacert>> |a valid filesystem path|No
 | <<plugins-{type}s-{plugin}-cloud_auth>> |<<password,password>>|No
@@ -324,6 +328,16 @@ The Elasticsearch action to perform. Valid actions are:
 
 For more details on actions, check out the http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html[Elasticsearch bulk API documentation]
 
+[id="plugins-{type}s-{plugin}-api_key"]
+===== `api_key`
+
+  * Value type is <<password,password>>
+  * There is no default value for this setting.
+
+Authenticate using Elasticsearch API key. Note that this option also requires enabling the `ssl` option.
+
+Format is `id:api_key` where `id` and `api_key` are as returned by the Elasticsearch https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html[Create API key API].
+
 [id="plugins-{type}s-{plugin}-bulk_path"]
 ===== `bulk_path` 
 

lib/logstash/outputs/elasticsearch.rb

@@ -122,6 +122,10 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # Password to authenticate to a secure Elasticsearch cluster
   config :password, :validate => :password
 
+  # Authenticate using Elasticsearch API key.
+  # format is id:api_key (as returned by https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html[Create API key])
+  config :api_key, :validate => :password
+
   # Cloud authentication string ("<username>:<password>" format) is an alternative for the `user`/`password` configuration.
   #
   # For more details, check out the https://www.elastic.co/guide/en/logstash/current/connecting-to-cloud.html#_cloud_auth[cloud documentation]
@@ -255,6 +259,14 @@ def config_init(params)
   end
 
   def build_client
+    # the following 3 options validation & setup methods are called inside build_client
+    # because they must be executed prior to building the client and logstash
+    # monitoring and management rely on directly calling build_client
+    # see https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/934#pullrequestreview-396203307
+    validate_authentication
+    fill_hosts_from_cloud_id
+    setup_hosts
+
     params["metric"] = metric
     if @proxy.eql?('')
       @logger.warn "Supplied proxy setting (proxy => '') has no effect"

lib/logstash/outputs/elasticsearch/common.rb

@@ -20,10 +20,6 @@ def register
       @stopping = Concurrent::AtomicBoolean.new(false)
       # To support BWC, we check if DLQ exists in core (< 5.4). If it doesn't, we use nil to resort to previous behavior.
       @dlq_writer = dlq_enabled? ? execution_context.dlq_writer : nil
-
-      fill_hosts_from_cloud_id
-      fill_user_password_from_cloud_auth
-      setup_hosts # properly sets @hosts
       build_client
       setup_after_successful_connection
       check_action_validity
@@ -112,6 +108,28 @@ def event_action_tuple(event)
       [action, params, event]
     end
 
+    def validate_authentication
+      authn_options = 0
+      authn_options += 1 if @cloud_auth
+      authn_options += 1 if (@api_key && @api_key.value)
+      authn_options += 1 if (@user || (@password && @password.value))
+
+      if authn_options > 1
+        raise LogStash::ConfigurationError, 'Multiple authentication options are specified, please only use one of user/password, cloud_auth or api_key'
+      end
+
+      if @api_key && @api_key.value && @ssl   != true
+        raise(LogStash::ConfigurationError, "Using api_key authentication requires SSL/TLS secured communication using the `ssl => true` option")
+      end
+
+      if @cloud_auth
+        @user, @password = parse_user_password_from_cloud_auth(@cloud_auth)
+        # params is the plugin global params hash which will be passed to HttpClientBuilder.build
+        params['user'], params['password'] = @user, @password
+      end
+    end
+    private :validate_authentication
+
     def setup_hosts
       @hosts = Array(@hosts)
       if @hosts.empty?
@@ -135,16 +153,6 @@ def fill_hosts_from_cloud_id
       @hosts = parse_host_uri_from_cloud_id(@cloud_id)
     end
 
-    def fill_user_password_from_cloud_auth
-      return unless @cloud_auth
-
-      if @user || @password
-        raise LogStash::ConfigurationError, 'Both cloud_auth and user/password specified, please only use one.'
-      end
-      @user, @password = parse_user_password_from_cloud_auth(@cloud_auth)
-      params['user'], params['password'] = @user, @password
-    end
-
     def parse_host_uri_from_cloud_id(cloud_id)
       begin # might not be available on older LS
         require 'logstash/util/cloud_setting_id'

lib/logstash/outputs/elasticsearch/http_client_builder.rb

@@ -1,4 +1,5 @@
 require 'cgi'
+require "base64"
 
 module LogStash; module Outputs; class ElasticSearch;
   module HttpClientBuilder
@@ -8,7 +9,7 @@ def self.build(logger, hosts, params)
         :pool_max_per_route => params["pool_max_per_route"],
         :check_connection_timeout => params["validate_after_inactivity"],
         :http_compression => params["http_compression"],
-        :headers => params["custom_headers"]
+        :headers => params["custom_headers"] || {}
       }
 
       client_settings[:proxy] = params["proxy"] if params["proxy"]
@@ -56,6 +57,7 @@ def self.build(logger, hosts, params)
 
       client_settings.merge! setup_ssl(logger, params)
       common_options.merge! setup_basic_auth(logger, params)
+      client_settings[:headers].merge! setup_api_key(logger, params)
 
       external_version_types = ["external", "external_gt", "external_gte"]
       # External Version validation
@@ -151,6 +153,14 @@ def self.setup_basic_auth(logger, params)
       }
     end
 
+    def self.setup_api_key(logger, params)
+      api_key = params["api_key"]
+
+      return {} unless (api_key && api_key.value)
+
+      { "Authorization" => "ApiKey " + Base64.strict_encode64(api_key.value) }
+    end
+
     private
     def self.dedup_slashes(url)
       url.gsub(/\/+/, "/")

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '10.4.1'
+  s.version         = '10.5.0'
 
   s.licenses        = ['apache-2.0']
   s.summary         = "Stores logs in Elasticsearch"

spec/unit/outputs/elasticsearch_spec.rb

@@ -1,4 +1,5 @@
 require_relative "../../../spec/es_spec_helper"
+require "base64"
 require "flores/random"
 require "logstash/outputs/elasticsearch"
 
@@ -142,6 +143,25 @@
 
         include_examples("an authenticated config")
       end
+
+      context 'claud_auth also set' do
+        let(:do_register) { false } # this is what we want to test, so we disable the before(:each) call
+        let(:options) { { "user" => user, "password" => password, "cloud_auth" => "elastic:my-passwd-00" } }
+
+        it "should fail" do
+          expect { subject.register }.to raise_error LogStash::ConfigurationError, /Multiple authentication options are specified/
+        end
+      end
+
+      context 'api_key also set' do
+        let(:do_register) { false } # this is what we want to test, so we disable the before(:each) call
+        let(:options) { { "user" => user, "password" => password, "api_key" => "some_key" } }
+
+        it "should fail" do
+          expect { subject.register }.to raise_error LogStash::ConfigurationError, /Multiple authentication options are specified/
+        end
+      end
+
     end
 
     describe "with path" do
@@ -577,7 +597,15 @@
       let(:options) { { 'cloud_auth' => 'elastic:my-passwd-00', 'user' => 'another' } }
 
       it "should fail" do
-        expect { subject.register }.to raise_error LogStash::ConfigurationError, /cloud_auth and user/
+        expect { subject.register }.to raise_error LogStash::ConfigurationError, /Multiple authentication options are specified/
+      end
+    end
+
+    context 'api_key also set' do
+      let(:options) { { 'cloud_auth' => 'elastic:my-passwd-00', 'api_key' => 'some_key' } }
+
+      it "should fail" do
+        expect { subject.register }.to raise_error LogStash::ConfigurationError, /Multiple authentication options are specified/
       end
     end
   end if LOGSTASH_VERSION > '6.0'
@@ -659,6 +687,62 @@
     end
   end
 
+  describe "API key" do
+    let(:manticore_options) { subject.client.pool.adapter.manticore.instance_variable_get(:@options) }
+    let(:api_key) { "some_id:some_api_key" }
+    let(:base64_api_key) { "ApiKey c29tZV9pZDpzb21lX2FwaV9rZXk=" }
+
+    context "when set without ssl" do
+      let(:do_register) { false } # this is what we want to test, so we disable the before(:each) call
+      let(:options) { { "api_key" => api_key } }
+
+      it "should raise a configuration error" do
+        expect { subject.register }.to raise_error LogStash::ConfigurationError, /requires SSL\/TLS/
+      end
+    end
+
+    context "when set without ssl but with a https host" do
+      let(:do_register) { false } # this is what we want to test, so we disable the before(:each) call
+      let(:options) { { "hosts" => ["https://some.host.com"], "api_key" => api_key } }
+
+      it "should raise a configuration error" do
+        expect { subject.register }.to raise_error LogStash::ConfigurationError, /requires SSL\/TLS/
+      end
+    end
+
+    context "when set" do
+      let(:options) { { "ssl" => true, "api_key" =>  ::LogStash::Util::Password.new(api_key) } }
+
+      it "should use the custom headers in the adapter options" do
+        expect(manticore_options[:headers]).to eq({ "Authorization" => base64_api_key })
+      end
+    end
+
+    context "when not set" do
+      it "should have no headers" do
+        expect(manticore_options[:headers]).to be_empty
+      end
+    end
+
+    context 'user also set' do
+      let(:do_register) { false } # this is what we want to test, so we disable the before(:each) call
+      let(:options) { { "ssl" => true, "api_key" => api_key, 'user' => 'another' } }
+
+      it "should fail" do
+        expect { subject.register }.to raise_error LogStash::ConfigurationError, /Multiple authentication options are specified/
+      end
+    end
+
+    context 'cloud_auth also set' do
+      let(:do_register) { false } # this is what we want to test, so we disable the before(:each) call
+      let(:options) { { "ssl" => true, "api_key" => api_key, 'cloud_auth' => 'foobar' } }
+
+      it "should fail" do
+        expect { subject.register }.to raise_error LogStash::ConfigurationError, /Multiple authentication options are specified/
+      end
+    end
+  end
+
   @private
 
   def stub_manticore_client!(manticore_double = nil)