Improve URL parsing in 'hosts' to take a full URL as well as warn on …

…more errors

Fixes elastic#363

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 2.5.0
+ - Host settings now are more robust to bad input
+ - Host settings can now take full URLs
+
 ## 2.4.2
  - Make flush_size actually cap the batch size in LS 2.2+
 

lib/logstash/outputs/elasticsearch.rb

@@ -81,8 +81,10 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # the root path for the Elasticsearch HTTP API lives.
   config :path, :validate => :string, :default => "/"
 
-  # Enable SSL/TLS secured communication to Elasticsearch cluster
-  config :ssl, :validate => :boolean, :default => false
+  # Enable SSL/TLS secured communication to Elasticsearch cluster. Leaving this unspecified will use whatever scheme
+  # is specified in the URLs listed in 'hosts'. If no explicit protocol is specified plain HTTP will be used.
+  # If SSL is explicitly disabled here the plugin will refuse to start if an HTTPS URL is given in 'hosts'
+  config :ssl, :validate => :boolean
 
   # Option to validate the server's certificate. Disabling this severely compromises security.
   # For more information on disabling certificate verification please read

lib/logstash/outputs/elasticsearch/common_configs.rb

@@ -74,6 +74,9 @@ def self.included(mod)
       # Remember the `http` protocol uses the http://www.elastic.co/guide/en/elasticsearch/reference/current/modules-http.html#modules-http[http] address (eg. 9200, not 9300).
       #     `"127.0.0.1"`
       #     `["127.0.0.1:9200","127.0.0.2:9200"]`
+      #     `["http://127.0.0.1"]`
+      #     `["https://127.0.0.1:9200"]`
+      #     `["https://127.0.0.1:9200/mypath"]` (If using a proxy on a subpath)
       # It is important to exclude http://www.elastic.co/guide/en/elasticsearch/reference/current/modules-node.html[dedicated master nodes] from the `hosts` list
       # to prevent LS from sending bulk requests to the master nodes.  So this parameter should only reference either data or client nodes in Elasticsearch.
       mod.config :hosts, :validate => :array, :default => ["127.0.0.1"]

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -105,18 +105,10 @@ def build_client(options)
       client_settings = options[:client_settings] || {}
       timeout = options[:timeout] || 0
 
-      uris = hosts.map do |host|
-        proto = client_settings[:ssl] ? "https"  : "http"
-        if host =~ /:\d+\z/
-          "#{proto}://#{host}#{client_settings[:path]}"
-        else
-          # Use default port of 9200 if none provided with host.
-          "#{proto}://#{host}:9200#{client_settings[:path]}"
-        end
-      end
+      urls = hosts.map {|host| host_to_url(host, client_settings[:ssl], client_settings[:path])}
 
       @client_options = {
-        :hosts => uris,
+        :hosts => urls,
         :ssl => client_settings[:ssl],
         :transport_options => {
           :socket_timeout => timeout,
@@ -136,6 +128,61 @@ def build_client(options)
       Elasticsearch::Client.new(client_options)
     end
 
+    HOSTNAME_PORT_REGEX=/\A(?<hostname>([A-Za-z0-9\.\-]+)|\[[0-9A-Fa-f\:]+\])(:(?<port>\d+))?\Z/
+    URL_REGEX=/\A#{URI::regexp(['http', 'https'])}\z/
+    # Parse a configuration host to a normalized URL
+    def host_to_url(host, ssl=nil, path=nil)
+      explicit_scheme = case ssl
+                        when true
+                          "https"
+                        when false
+                          "http"
+                        else
+                          nil
+                        end
+
+      # Ensure path starts with a /
+      if path && path[0] != '/'
+        path = "/#{path}"
+      end
+
+      url = nil
+      if host =~ URL_REGEX
+        url = URI.parse(host)
+
+        # Please note that the ssl == nil case is different! If you didn't make an explicit
+        # choice we don't complain!
+        if url.scheme == "http" && ssl == true
+          raise LogStash::ConfigurationError, "You specified a plain 'http' URL '#{host}' but set 'ssl' to true! Aborting!"
+        elsif url.scheme == "https" && ssl == false
+          raise LogStash::ConfigurationError, "You have explicitly disabled SSL but passed in an https URL '#{host}'! Aborting!"
+        end
+
+        url.scheme = explicit_scheme if explicit_scheme
+      elsif (match_results = HOSTNAME_PORT_REGEX.match(host))
+        hostname = match_results["hostname"]
+        port = match_results["port"] || 9200
+        url = URI.parse("#{explicit_scheme || 'http'}://#{hostname}:#{port}")
+      else
+        raise LogStash::ConfigurationError, "Host '#{host}' was specified, but is not valid! Use either a full URL or a hostname:port string!"
+      end
+
+      if path && url.path && url.path != "/" && url.path != ''
+        raise LogStash::ConfigurationError, "A path '#{url.path}' has been explicitly specified in the url '#{url}', but you also specified a path of '#{path}'. This is probably a mistake, please remove one setting."
+      end
+
+      if path
+        url.path = path  # The URI library cannot stringify if it holds a nil
+      end
+
+      if url.password || url.user
+        raise LogStash::ConfigurationError, "We do not support setting the user password in the URL directly as " +
+          "this may be logged to disk thus leaking credentials. Use the 'user' and 'password' options respectively"
+      end
+
+      url.to_s
+    end
+
     def template_exists?(name)
       @client.indices.get_template(:name => name)
       return true

logstash-output-elasticsearch.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '2.4.2'
+  s.version         = '2.5.0'
   s.licenses        = ['apache-2.0']
   s.summary         = "Logstash Output to Elasticsearch"
   s.description     = "Output events to elasticsearch"

spec/unit/outputs/elasticsearch/http_client_spec.rb

@@ -0,0 +1,129 @@
+require "logstash/devutils/rspec/spec_helper"
+require "logstash/outputs/elasticsearch/http_client"
+require "java"
+
+describe LogStash::Outputs::ElasticSearch::HttpClient do
+  let(:base_options) { {:hosts => ["127.0.0.1"], :logger => Cabin::Channel.get }}
+
+  describe "Host/URL Parsing" do
+    subject { described_class.new(base_options) }
+
+    let(:true_hostname) { "my-dash.hostname" }
+    let(:ipv6_hostname) { "[::1]" }
+    let(:ipv4_hostname) { "127.0.0.1" }
+    let(:port) { 9202 }
+    let(:hostname_port) { "#{hostname}:#{port}"}
+    let(:http_hostname_port) { "http://#{hostname_port}"}
+    let(:https_hostname_port) { "https://#{hostname_port}"}
+    let(:http_hostname_port_path) { "http://#{hostname_port}/path"}
+
+    shared_examples("proper host handling") do
+      it "should properly transform a host:port string to a URL" do
+        expect(subject.send(:host_to_url, hostname_port)).to eql(http_hostname_port)
+      end
+
+      it "should raise an error when a partial URL is an invalid format" do
+        expect {
+          subject.send(:host_to_url, "#{hostname_port}/")
+        }.to raise_error(LogStash::ConfigurationError)
+      end
+
+      it "should not raise an error with a / for a path" do
+        expect(subject.send(:host_to_url, "#{http_hostname_port}/")).to eql("#{http_hostname_port}/")
+      end
+
+      it "should parse full URLs correctly" do
+        expect(subject.send(:host_to_url, http_hostname_port)).to eql(http_hostname_port)
+      end
+
+      it "should reject full URLs with usernames and passwords" do
+        expect {
+          subject.send(:host_to_url, "http://user:password@host.domain")
+        }.to raise_error(LogStash::ConfigurationError)
+      end
+
+      describe "ssl" do
+        it "should refuse to handle an http url when ssl is true" do
+          expect {
+            subject.send(:host_to_url, http_hostname_port, true)
+          }.to raise_error(LogStash::ConfigurationError)
+        end
+
+        it "should refuse to handle an https url when ssl is false" do
+          expect {
+            subject.send(:host_to_url, https_hostname_port, false)
+          }.to raise_error(LogStash::ConfigurationError)
+        end
+
+        it "should handle an ssl url correctly when SSL is nil" do
+          expect(subject.send(:host_to_url, https_hostname_port, nil)).to eql(https_hostname_port)
+        end
+      end
+
+      describe "path" do
+        it "should allow paths in a url" do
+          expect(subject.send(:host_to_url, http_hostname_port_path, nil)).to eql(http_hostname_port_path)
+        end
+
+        it "should not allow paths in two places" do
+          expect {
+            subject.send(:host_to_url, http_hostname_port_path, false, "/otherpath")
+          }.to raise_error(LogStash::ConfigurationError)
+        end
+
+        it "should automatically insert a / in front of path overlays if needed" do
+          expect(subject.send(:host_to_url, http_hostname_port, false, "otherpath")).to eql(http_hostname_port + "/otherpath")
+        end
+      end
+    end
+
+    describe "an regular hostname" do
+      let(:hostname) { true_hostname }
+      include_examples("proper host handling")
+    end
+
+    describe "an ipv4 host" do
+      let(:hostname) { ipv4_hostname }
+      include_examples("proper host handling")
+    end
+
+    describe "an ipv6 host" do
+      let(:hostname) { ipv6_hostname }
+      include_examples("proper host handling")
+    end
+  end
+
+  describe "sniffing" do
+    let(:client) { LogStash::Outputs::ElasticSearch::HttpClient.new(base_options.merge(client_opts)) }
+    let(:transport) { client.client.transport }
+
+    before do
+      allow(transport).to receive(:reload_connections!)
+    end
+
+    context "with sniffing enabled" do
+      let(:client_opts) { {:sniffing => true, :sniffing_delay => 1 } }
+
+      after do
+        client.stop_sniffing!
+      end
+
+      it "should start the sniffer" do
+        expect(client.sniffer_thread).to be_a(Thread)
+      end
+
+      it "should periodically sniff the client" do
+        sleep 2
+        expect(transport).to have_received(:reload_connections!).at_least(:once)
+      end
+    end
+
+    context "with sniffing disabled" do
+      let(:client_opts) { {:sniffing => false} }
+
+      it "should not start the sniffer" do
+        expect(client.sniffer_thread).to be_nil
+      end
+    end
+  end
+end