out_opensearch: limit what a host placeholder can expand to

lib/fluent/plugin/out_opensearch.rb

@@ -95,6 +95,16 @@ def initialize(retry_stream)
     DEFAULT_TARGET_BULK_BYTES = -1
     DEFAULT_POLICY_ID = "logstash-policy"
 
+    # A placeholder in `host`/`hosts` is expanded from the tag or a record
+    # field, so a log sender chooses its value. Without a check the value
+    # could add a host with ",", userinfo with "@" or a path with "/", and
+    # the configured user/password and custom_headers would go to that host.
+    HOST_PLACEHOLDER_PATTERN = /\$\{[^}]*\}/
+    HOST_LABEL_PATTERN = /\A[0-9A-Za-z_\-]+\z/
+    HOST_NAME_PATTERN = /\A[0-9A-Za-z_\-]+(?:\.[0-9A-Za-z_\-]+)*\z/
+    # A host name or an IPv6 address in "[]", and the port that may follow it.
+    HOST_AND_PORT_PATTERN = /\A(\[[^\]]+\]|[^:]*)(?::([0-9]+))?\z/
+
     config_param :host, :string,  :default => 'localhost'
     config_param :port, :integer, :default => 9200
     config_param :user, :string, :default => nil
@@ -316,6 +326,8 @@ def configure(conf)
         log.info "host placeholder and template installation makes your OpenSearch cluster a bit slow down(beta)."
       end
 
+      warn_host_placeholder_with_credentials
+
       @template_names = []
       if !dry_run?
         if @template_name && @template_file
@@ -481,6 +493,34 @@ def dry_run?
       end
     end
 
+    # A placeholder decides the host, so every credential the configuration
+    # holds follows it. `endpoint` is left alone because
+    # `get_connection_options` ignores the expanded host there.
+    def warn_host_placeholder_with_credentials
+      return if @endpoint
+
+      elements = (@hosts || @host).split(',')
+      return unless elements.any? { |element| element.match?(HOST_PLACEHOLDER_PATTERN) }
+
+      param_name = @hosts ? 'hosts' : 'host'
+      if @user || @password || !@custom_headers.empty?
+        log.warn "'#{param_name}' uses a placeholder, so 'user', 'password' and 'custom_headers' are sent to whichever host a tag or a record field expands to. Make sure that only trusted senders can set them."
+      end
+      if elements.any? { |element| credentials_before_placeholder?(element) }
+        # Do not put the element in the log: it might holds the password.
+        log.warn "'#{param_name}' has a user and password in front of a host that a placeholder decides, so they are sent to whichever host a tag or a record field expands to. Make sure that only trusted senders can set it."
+      end
+    end
+
+    # A user and password written in a URL sit in front of the host, so a
+    # placeholder that comes after them picks where they are sent.
+    def credentials_before_placeholder?(element)
+      userinfo = element.index('@')
+      placeholder = element.index('${')
+
+      !userinfo.nil? && !placeholder.nil? && userinfo < placeholder
+    end
+
     def placeholder?(name, param)
       placeholder_validities = []
       placeholder_validators(name, param).each do |v|
@@ -847,6 +887,52 @@ def expand_placeholders(chunk)
       return logstash_prefix, logstash_dateformat, index_name, template_name, customize_template, application_name, pipeline
     end
 
+    def expand_host_placeholders(chunk)
+      template = @hosts || @host
+      template.split(',').each do |element|
+        offset = 0
+        while (match = HOST_PLACEHOLDER_PATTERN.match(element, offset))
+          offset = match.end(0)
+          value = extract_placeholders(match[0], chunk)
+          next if valid_host_placeholder_value?(value, match.pre_match, match.post_match)
+
+          raise UnrecoverableRequestFailure, "Rejected #{value.dump} for #{match[0]} in 'host'/'hosts': a placeholder may only fill in the part of a host name that it stands for."
+        end
+      end
+
+      extract_placeholders(template, chunk)
+    end
+
+    # What a value may hold depends on where the operator put the placeholder.
+    # `before` and `after` are the parts of the element around it.
+    def valid_host_placeholder_value?(value, before, after)
+      # The operator wrote the "[]" that an IPv6 address needs.
+      return Resolv::IPv6::Regex.match?(value) if before.end_with?('[') && after.start_with?(']')
+
+      # The value continues a name the operator wrote, so a "." would move the
+      # request to another domain and a port cannot follow it.
+      return HOST_LABEL_PATTERN.match?(value) unless whole_host_placeholder?(before, after)
+
+      host, port = value.match(HOST_AND_PORT_PATTERN)&.captures
+      # The value may end with a port, unless the operator wrote one already.
+      return false if host.nil? || (port && after.start_with?(':'))
+
+      HOST_NAME_PATTERN.match?(host) || bracketed_ipv6?(host, before)
+    end
+
+    # The value fills in the whole host name when only a scheme or a userinfo
+    # comes before it and only a port or a path comes after it.
+    def whole_host_placeholder?(before, after)
+      (before.empty? || before.end_with?('//', '@')) &&
+        (after.empty? || after.start_with?(':', '/', '?', '#'))
+    end
+
+    # `URI()` in `get_connection_options` needs a scheme to read an IPv6
+    # address, and the operator writes the scheme, not the value.
+    def bracketed_ipv6?(host, before)
+      !before.empty? && host.start_with?('[') && Resolv::IPv6::Regex.match?(host[1..-2])
+    end
+
     def multi_workers_ready?
       true
     end
@@ -869,11 +955,7 @@ def write(chunk)
       tag = chunk.metadata.tag
       chunk_id = dump_unique_id_hex(chunk.unique_id)
       extracted_values = expand_placeholders(chunk)
-      host = if @hosts
-               extract_placeholders(@hosts, chunk)
-             else
-               extract_placeholders(@host, chunk)
-             end
+      host = expand_host_placeholders(chunk)
 
       affinity_target_indices = get_affinity_target_indices(chunk)
       chunk.msgpack_each do |time, record|

lib/fluent/plugin/out_opensearch_data_stream.rb

@@ -155,11 +155,7 @@ def write(chunk)
       data_stream_template_name = @data_stream_template_name
       host = nil
       if @use_placeholder
-        host = if @hosts
-                 extract_placeholders(@hosts, chunk)
-               else
-                 extract_placeholders(@host, chunk)
-               end
+        host = expand_host_placeholders(chunk)
         data_stream_name = extract_placeholders(@data_stream_name, chunk).downcase
         data_stream_template_name = extract_placeholders(@data_stream_template_name, chunk).downcase
         begin

test/plugin/test_out_opensearch.rb

@@ -2876,6 +2876,114 @@ def test_writes_to_extracted_host_with_placeholder_replaced_in_exception_message
       connection_spec = {host: "myhost-1", port: 9200, scheme: "http"}
       assert_equal("could not push logs to OpenSearch cluster (#{connection_spec.inspect}): [503] ", exception.message)
     end
+
+    # `extract_placeholders` needs a real chunk, so replace it with the value
+    # that a crafted tag or record field would expand to.
+    def expand_hosts_with(hosts, placeholder_value)
+      instance = driver.configure("hosts #{hosts}").instance
+      instance.define_singleton_method(:extract_placeholders) do |value, _chunk|
+        value.gsub('${pipeline_id}', placeholder_value)
+      end
+      instance.expand_host_placeholders(nil)
+    end
+
+    data("host name"   => ["${pipeline_id},logs2.example.com:9201", "os.internal"],
+         "with port"   => ["${pipeline_id},logs2.example.com:9201", "os.internal:9200"],
+         "in a url"    => ["https://${pipeline_id}/os", "logs.example.com:9201"],
+         "ipv6"        => ["http://${pipeline_id}:9201", "[2001:db8::1]"],
+         "ipv6 + port" => ["http://${pipeline_id}/os", "[2001:db8::1]:9201"],
+         "ipv6 in []"  => ["http://[${pipeline_id}]:9201", "2001:db8::1"])
+    def test_accepts_placeholder_value_which_fills_in_the_whole_host(data)
+      hosts, placeholder_value = data
+      assert_equal(hosts.sub('${pipeline_id}', placeholder_value),
+                   expand_hosts_with(hosts, placeholder_value))
+    end
+
+    data("extra host"   => "evil.example.com,myhost",
+         "userinfo"     => "1@evil.example.com",
+         "path"         => "1/evil.example.com",
+         "space"        => "1 evil.example.com",
+         "other domain" => "evil.example.net",
+         "port"         => "1:9999",
+         "empty"        => "")
+    def test_rejects_placeholder_value_which_does_not_fit_its_place(placeholder_value)
+      assert_raise(Fluent::Plugin::OpenSearchOutput::UnrecoverableRequestFailure) {
+        expand_hosts_with("myhost-${pipeline_id},logs2.example.com:9201", placeholder_value)
+      }
+    end
+
+    # `get_connection_options` reads an element with `URI()`, which cannot
+    # read an IPv6 address when the element has no scheme.
+    data("empty"     => "",
+         "port only" => ":9999",
+         "two ports" => "os.internal:9200:9201",
+         "path"      => "myhost/evil.example.com",
+         "ipv6"      => "[2001:db8::1]")
+    def test_rejects_placeholder_value_which_is_not_a_host_name(placeholder_value)
+      assert_raise(Fluent::Plugin::OpenSearchOutput::UnrecoverableRequestFailure) {
+        expand_hosts_with("${pipeline_id},logs2.example.com:9201", placeholder_value)
+      }
+    end
+
+    data("host name" => "os.internal:9200",
+         "ipv6"      => "[2001:db8::1]:9200")
+    def test_rejects_placeholder_value_with_a_port_when_the_port_is_configured(placeholder_value)
+      assert_raise(Fluent::Plugin::OpenSearchOutput::UnrecoverableRequestFailure) {
+        expand_hosts_with("http://${pipeline_id}:9201", placeholder_value)
+      }
+    end
+
+    data("in []"      => ["http://[${pipeline_id}]:9201", "2001:db8"],
+         "extra host" => ["http://[${pipeline_id}]:9201", "[2001:db8::1],[2001:db8::2]"],
+         "whole host" => ["http://${pipeline_id}:9201", "[2001:db8]"])
+    def test_rejects_placeholder_value_which_is_not_an_ipv6_address(data)
+      hosts, placeholder_value = data
+      assert_raise(Fluent::Plugin::OpenSearchOutput::UnrecoverableRequestFailure) {
+        expand_hosts_with(hosts, placeholder_value)
+      }
+    end
+
+    # The operator writes the structure, so only the placeholder value is
+    # checked. Every form that `get_connection_options` accepts keeps working.
+    data("plain host" => ["myhost-${pipeline_id},logs2.example.com:9201",
+                          "myhost-1,logs2.example.com:9201"],
+         "url"        => ["https://logs-${pipeline_id}.example.com:9201/os",
+                          "https://logs-1.example.com:9201/os"],
+         "userinfo"   => ["https://john:pass@logs-${pipeline_id}.example.com/os",
+                          "https://john:pass@logs-1.example.com/os"],
+         "ipv6"       => ["http://[2001:db8::1]:9201,myhost-${pipeline_id}",
+                          "http://[2001:db8::1]:9201,myhost-1"])
+    def test_keeps_the_configured_host_structure(data)
+      hosts, expected = data
+      assert_equal(expected, expand_hosts_with(hosts, "1"))
+    end
+
+    def test_warns_when_placeholder_host_is_used_with_credentials
+      d = driver(%{
+        host     logs-${tag}.example.com
+        user     john
+        password doe
+        @log_level info
+      })
+      assert_true(d.logs.any? { |log| log.include?("'host' uses a placeholder") })
+    end
+
+    def test_warns_when_hosts_embeds_credentials_in_front_of_a_placeholder
+      d = driver(%{
+        hosts https://john:secret@logs-${tag}.example.com/os
+        @log_level info
+      })
+      assert_true(d.logs.any? { |log| log.include?("in front of a host that a placeholder decides") })
+      assert_false(d.logs.any? { |log| log.include?("secret") })
+    end
+
+    def test_does_not_warn_when_credentials_belong_to_a_static_host
+      d = driver(%{
+        hosts https://john:secret@static.example.com/os,logs-${tag}.example.com
+        @log_level info
+      })
+      assert_false(d.logs.any? { |log| log.include?("in front of a host that a placeholder decides") })
+    end
   end
 
   def test_writes_to_logstash_index_with_specified_prefix_uppercase