diff --git a/lib/fluent/plugin/out_opensearch.rb b/lib/fluent/plugin/out_opensearch.rb index 6755a57..84dda14 100644 --- a/lib/fluent/plugin/out_opensearch.rb +++ b/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| diff --git a/lib/fluent/plugin/out_opensearch_data_stream.rb b/lib/fluent/plugin/out_opensearch_data_stream.rb index 0518019..92389e7 100644 --- a/lib/fluent/plugin/out_opensearch_data_stream.rb +++ b/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 diff --git a/test/plugin/test_out_opensearch.rb b/test/plugin/test_out_opensearch.rb index 2107465..56fc128 100644 --- a/test/plugin/test_out_opensearch.rb +++ b/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