make ssl verification mode configurable

docs/index.asciidoc

@@ -99,6 +99,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-size>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-slices>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-ssl>> |<<boolean,boolean>>|No
+| <<plugins-{type}s-{plugin}-ssl_certificate_verification>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-user>> |<<string,string>>|No
 |=======================================================================
 
@@ -112,6 +113,7 @@ input plugins.
 
   * Value type is <<path,path>>
   * There is no default value for this setting.
+  * This option is ignored when <<plugins-{type}s-{plugin}-ssl>> is disabled.
 
 SSL Certificate Authority file in PEM encoded format, must also
 include any chain certificates as necessary.
@@ -284,6 +286,19 @@ instructions into the query.
 If enabled, SSL will be used when communicating with the Elasticsearch
 server (i.e. HTTPS will be used instead of plain HTTP).
 
+[id="plugins-{type}s-{plugin}-ssl_certificate_verification"]
+===== `ssl_certificate_verification`
+
+  * Value type is <<boolean,boolean>>
+  * Default value is `true`
+  * This option is ignored when <<plugins-{type}s-{plugin}-ssl>> is disabled.
+
+Disabling SSL verification is UNSAFE, and causes this input plugin to implicitly trust
+any server it connects to, including servers that present invalid, expired, forged, or
+self-signed certificates. When this setting is disabled, the connection between this
+input and the server that responds will be encrypted, but the identity of the server
+will not be validated.
+
 [id="plugins-{type}s-{plugin}-user"]
 ===== `user` 
 

lib/logstash/inputs/elasticsearch.rb

@@ -139,6 +139,11 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   # SSL Certificate Authority file in PEM encoded format, must also include any chain certificates as necessary 
   config :ca_file, :validate => :path
 
+  # SSL Certificate Verification is on by default.
+  # Disabling it is UNSAFE and means that we will trust any server that we connect to,
+  # including servers with invalid, expired, forged, or self-signed certificates.
+  config :ssl_certificate_verification, :validate => :boolean, :default => true
+
   # Schedule of when to periodically run statement, in Cron format
   # for example: "* * * * *" (execute query every minute, on the minute)
   #
@@ -177,8 +182,20 @@ def register
       @hosts
     end
 
-    if @ssl && @ca_file
-      transport_options[:ssl] = { :ca_file => @ca_file }
+    if @ssl
+      ssl_options = {:enable => true}
+      ssl_options[:ca_file] = @ca_file unless @ca_file.nil?
+      ssl_options[:verify] = @ssl_certificate_verification
+
+      if !@ssl_certificate_verification
+        logger.warn([
+                        "** WARNING ** Detected UNSAFE options in elasticsearch input configuration!",
+                        "** WARNING ** You have enabled encryption but DISABLED certificate verification.",
+                        "** WARNING ** To make sure your data is secure change :ssl_certificate_verification to true"
+                    ].join("\n"))
+      end
+
+      transport_options[:ssl] = ssl_options
     end
 
     @client = Elasticsearch::Client.new(:hosts => hosts, :transport_options => transport_options)

spec/inputs/elasticsearch_spec.rb

@@ -577,4 +577,104 @@ def synchronize_method!(object, method_name)
 
   end
 
+  context 'elasticsearch client configuration' do
+    # In this set of tests, we validate the client that is created by registering a plugin
+    # with a certain configuration, in order to ensure that certain client-specific settings
+    # are wired through correctly.
+    subject(:client_initialize_options) do
+      result = nil
+
+      expect(Elasticsearch::Client).to receive(:new) do |options|
+        result = options
+      end
+      plugin.register
+
+      # Ensure that our expectation on Elasticsearch::Client#new has been met
+      RSpec::Mocks.verify
+
+      result
+    end
+
+    let(:config) do
+      {
+          "hosts" => ["localhost:9200"],
+          "query" => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
+      }
+    end
+
+    context 'when `ssl` is explicitly enabled' do
+      let(:config) { super().merge('ssl' => 'true') }
+
+      it { should include(:hosts) }
+      context '-> :hosts' do
+        subject(:hosts) { client_initialize_options.fetch(:hosts) }
+        it 'maps each host to include https schema' do
+          expect(hosts).to eq([{host: 'localhost', scheme: 'https', port: '9200'}])
+        end
+      end
+
+      it { should include(:transport_options) }
+      context('-> :transport_options') do
+        subject(:transport_options) { client_initialize_options.fetch(:transport_options) }
+
+        it { should include(:ssl) }
+        context('-> :ssl') do
+          subject(:ssl_options) { transport_options.fetch(:ssl) }
+          it 'includes `enable: true`' do
+            expect(ssl_options).to include(:enable)
+            expect(ssl_options[:enable]).to be true
+          end
+          context('when `ssl_certificate_verification` is not specified') do
+            it 'includes `verify: true`' do
+              expect(ssl_options).to include(:verify)
+              expect(ssl_options[:verify]).to be true
+            end
+          end
+          context('when `ssl_certificate_verification` is explicitly enabled') do
+            let(:config) { super().merge('ssl_certificate_verification' => true) }
+            it 'includes `verify: true`' do
+              expect(ssl_options).to include(:verify)
+              expect(ssl_options[:verify]).to be true
+            end
+          end
+          context('when `ssl_certificate_verification` is explicitly disabled') do
+            let(:config) { super().merge('ssl_certificate_verification' => false) }
+            it 'includes `verify: false`' do
+              expect(ssl_options).to include(:verify)
+              expect(ssl_options[:verify]).to be false
+            end
+          end
+        end
+      end
+    end
+
+    {
+        'not specified' => {},
+        'explicitly disabled' => { 'ssl' => 'false' }
+    }.each do |ssl_option_desc, ssl_option_config_override|
+      context "when `ssl` is #{ssl_option_desc}" do
+        let(:config) { super().merge(ssl_option_config_override) }
+
+        it { should include(:hosts) }
+        context '-> :hosts' do
+          subject(:hosts) { client_initialize_options.fetch(:hosts) }
+          it 'does not maps each host to include https schema' do
+            expect(hosts).to eq(config['hosts'])
+          end
+        end
+
+        it { should include(:transport_options) }
+        context('-> :transport_options') do
+          subject(:transport_options) { client_initialize_options.fetch(:transport_options) }
+
+          context('-> :ssl') do
+            subject(:ssl_options) { transport_options.fetch(:ssl, {}) }
+            it 'does not include `enable: true`' do
+              expect(ssl_options[:enable]).to_not be true
+            end
+          end
+        end
+      end
+    end
+  end
 end