From fe1864d6ef9fad6b1b280da94a07e965d22561e0 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Wed, 16 Sep 2026 06:05:19 +0000 Subject: [PATCH 1/2] out_opensearch: limit what a host placeholder can expand to Before: A placeholder in `host`/`hosts` was expanded and passed to the connection options as is. So a tag or a record field could add a host with ",", userinfo with "@" or a path with "/". The configured user, password and custom_headers were then sent to that host. After: Each placeholder value is checked before the whole setting is expanded, so a value can only fill in a host name and a port. The rest of the setting, such as the "," between hosts or a scheme, userinfo and path the operator wrote, is left as it is. A placeholder used together with credentials logs a warning at startup. Co-Authored-By: Claude Signed-off-by: Kentaro Hayashi --- lib/fluent/plugin/out_opensearch.rb | 58 +++++++++++++-- .../plugin/out_opensearch_data_stream.rb | 6 +- test/plugin/test_out_opensearch.rb | 70 +++++++++++++++++++ 3 files changed, 124 insertions(+), 10 deletions(-) diff --git a/lib/fluent/plugin/out_opensearch.rb b/lib/fluent/plugin/out_opensearch.rb index 6755a57..b3b0aac 100644 --- a/lib/fluent/plugin/out_opensearch.rb +++ b/lib/fluent/plugin/out_opensearch.rb @@ -95,6 +95,14 @@ def initialize(retry_stream) DEFAULT_TARGET_BULK_BYTES = -1 DEFAULT_POLICY_ID = "logstash-policy" + HOST_PLACEHOLDER_PATTERN = /\$\{[^}]*\}/ + # A placeholder in `host`/`hosts` is expanded from the tag or a record + # field, so its value may only fill in a host name and a port. + # without this a log sender could add a host with ",", userinfo with "@" or a path with "/", + # and receive the configured user/password and custom_headers. A port is + # allowed because it can only be chosen on a host the value already picked. + HOST_PLACEHOLDER_VALUE_PATTERN = /\A[0-9A-Za-z_\-.]*(:[0-9]+)?\z/ + config_param :host, :string, :default => 'localhost' config_param :port, :integer, :default => 9200 config_param :user, :string, :default => nil @@ -316,6 +324,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 +491,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 +885,20 @@ 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 + # Check each placeholder value, reject "," and so on here + # because it might change sending target host by sender unexpectedly. + template.scan(HOST_PLACEHOLDER_PATTERN).each do |placeholder| + value = extract_placeholders(placeholder, chunk) + next if HOST_PLACEHOLDER_VALUE_PATTERN.match?(value) + + raise UnrecoverableRequestFailure, "Rejected #{value.dump} for #{placeholder} in 'host'/'hosts': a placeholder may only fill in a host name and a port." + end + + extract_placeholders(template, chunk) + end + def multi_workers_ready? true end @@ -869,11 +921,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..253b874 100644 --- a/test/plugin/test_out_opensearch.rb +++ b/test/plugin/test_out_opensearch.rb @@ -2876,6 +2876,76 @@ 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("extra host" => "evil.example.com,myhost", + "userinfo" => "1@evil.example.com", + "path" => "1/evil.example.com", + "space" => "1 evil.example.com", + "two ports" => "1:9999:8888") + def test_rejects_placeholder_value_which_is_not_a_host_name(placeholder_value) + assert_raise(Fluent::Plugin::OpenSearchOutput::UnrecoverableRequestFailure) { + expand_hosts_with("myhost-${pipeline_id},logs2.example.com:9201", placeholder_value) + } + end + + # A port is allowed, so that a tag or a record field can carry + # "host:port". It only picks a port on a host the value already picked. + def test_accepts_placeholder_value_with_a_port + assert_equal("os.internal:9200,logs2.example.com:9201", + expand_hosts_with("${pipeline_id},logs2.example.com:9201", "os.internal:9200")) + 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://[2404:7a80:d440:3000:de:7311:6329:2e6c]:9201,myhost-${pipeline_id}", + "http://[2404:7a80:d440:3000:de:7311:6329:2e6c]: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 From 69ac590080497e13a2aeefc559d09dbeafcda3d3 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Thu, 17 Sep 2026 05:29:57 +0000 Subject: [PATCH 2/2] out_opensearch: check a host placeholder value by its place Before: a value could hold a host name and a port wherever the placeholder was. * `host myhost-${pipeline_id}` let a record field move the request to another domain * `hosts https://logs-${pipeline_id}.example.com/os` let a value put a port in the middle of a name, which `URI()` cannot read * An empty value dropped the host without an error * An IPv6 address was rejected everywhere After: what a value may hold depends on where the operator put the placeholder. * A value that continues a name the operator wrote fills in one label, so it can add neither a "." nor a port * A value that stands for the whole host name may hold a name and a port * An IPv6 address is allowed where the operator wrote the "[]" or the scheme that `URI()` needs to read it * An empty value is rejected everywhere Co-Authored-By: Claude Signed-off-by: Kentaro Hayashi --- lib/fluent/plugin/out_opensearch.rb | 60 +++++++++++++++++++++------ test/plugin/test_out_opensearch.rb | 64 +++++++++++++++++++++++------ 2 files changed, 98 insertions(+), 26 deletions(-) diff --git a/lib/fluent/plugin/out_opensearch.rb b/lib/fluent/plugin/out_opensearch.rb index b3b0aac..84dda14 100644 --- a/lib/fluent/plugin/out_opensearch.rb +++ b/lib/fluent/plugin/out_opensearch.rb @@ -95,13 +95,15 @@ def initialize(retry_stream) DEFAULT_TARGET_BULK_BYTES = -1 DEFAULT_POLICY_ID = "logstash-policy" - HOST_PLACEHOLDER_PATTERN = /\$\{[^}]*\}/ # A placeholder in `host`/`hosts` is expanded from the tag or a record - # field, so its value may only fill in a host name and a port. - # without this a log sender could add a host with ",", userinfo with "@" or a path with "/", - # and receive the configured user/password and custom_headers. A port is - # allowed because it can only be chosen on a host the value already picked. - HOST_PLACEHOLDER_VALUE_PATTERN = /\A[0-9A-Za-z_\-.]*(:[0-9]+)?\z/ + # 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 @@ -887,18 +889,50 @@ def expand_placeholders(chunk) def expand_host_placeholders(chunk) template = @hosts || @host - # Check each placeholder value, reject "," and so on here - # because it might change sending target host by sender unexpectedly. - template.scan(HOST_PLACEHOLDER_PATTERN).each do |placeholder| - value = extract_placeholders(placeholder, chunk) - next if HOST_PLACEHOLDER_VALUE_PATTERN.match?(value) - - raise UnrecoverableRequestFailure, "Rejected #{value.dump} for #{placeholder} in 'host'/'hosts': a placeholder may only fill in a host name and a port." + 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 diff --git a/test/plugin/test_out_opensearch.rb b/test/plugin/test_out_opensearch.rb index 253b874..56fc128 100644 --- a/test/plugin/test_out_opensearch.rb +++ b/test/plugin/test_out_opensearch.rb @@ -2887,22 +2887,60 @@ def expand_hosts_with(hosts, placeholder_value) instance.expand_host_placeholders(nil) end - data("extra host" => "evil.example.com,myhost", - "userinfo" => "1@evil.example.com", - "path" => "1/evil.example.com", - "space" => "1 evil.example.com", - "two ports" => "1:9999:8888") - def test_rejects_placeholder_value_which_is_not_a_host_name(placeholder_value) + 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 - # A port is allowed, so that a tag or a record field can carry - # "host:port". It only picks a port on a host the value already picked. - def test_accepts_placeholder_value_with_a_port - assert_equal("os.internal:9200,logs2.example.com:9201", - expand_hosts_with("${pipeline_id},logs2.example.com:9201", "os.internal:9200")) + # `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 @@ -2913,8 +2951,8 @@ def test_accepts_placeholder_value_with_a_port "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://[2404:7a80:d440:3000:de:7311:6329:2e6c]:9201,myhost-${pipeline_id}", - "http://[2404:7a80:d440:3000:de:7311:6329:2e6c]:9201,myhost-1"]) + "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"))