diff --git a/lib/fluent/plugin/in_http.rb b/lib/fluent/plugin/in_http.rb index c5df87ea3b..e3b94c58ae 100644 --- a/lib/fluent/plugin/in_http.rb +++ b/lib/fluent/plugin/in_http.rb @@ -24,6 +24,8 @@ require 'uri' require 'socket' require 'json' +require 'ipaddr' +require 'openssl' module Fluent::Plugin class InHttpParser < Parser @@ -90,6 +92,22 @@ class HttpInput < Input desc "Add prefix to incoming tag" config_param :add_tag_prefix, :string, default: nil + config_section :auth, required: false, multi: false do + desc 'The method for HTTP authentication' + config_param :method, :enum, list: [:basic], default: :basic + desc 'The username for basic authentication' + config_param :username, :string + desc 'The password for basic authentication' + config_param :password, :string, secret: true + end + + config_section :security, required: false, multi: false do + config_section :client, param_name: :clients, required: false, multi: true do + desc 'Address or network of the client, IPv4 or IPv6. e.g. 192.168.1.10, 192.168.0.0/16, fd00::/8' + config_param :network, :string + end + end + config_section :parse do config_set_default :@type, 'in_http' end @@ -100,6 +118,7 @@ def initialize super @km = nil + @nodes = [] @format_name = nil @parser_time_key = nil @@ -127,6 +146,8 @@ def configure(conf) raise Fluent::ConfigError, "'add_tag_prefix' parameter must not be empty" if @add_tag_prefix && @add_tag_prefix.empty? + configure_security if @security + m = if @parser_configs.first['@type'] == 'in_http' @parser_msgpack = parser_create(usage: 'parser_in_http_msgpack', type: 'msgpack') @parser_msgpack.time_key = nil @@ -260,11 +281,33 @@ def on_request(path_info, params) private + def configure_security + @nodes = @security.clients.map do |client| + begin + IPAddr.new(client.network) + rescue ArgumentError + raise Fluent::ConfigError, "network '#{client.network}' address format is invalid" + end + end + end + + def allowed_client?(address) + return true if @nodes.empty? + @nodes.any? { |node| node.include?(address) rescue false } + end + def on_server_connect(conn) + unless allowed_client?(conn.remote_addr) + log.warn "client address does not match any allowed network", address: conn.remote_addr + conn.data {|_data| } + conn.close + return + end + handler = Handler.new(conn, @km, method(:on_request), @body_size_limit, @decompression_size_limit, @format_name, log, @cors_allow_origins, @cors_allow_credentials, - @add_query_params) + @add_query_params, @auth) conn.on(:data) do |data| handler.on_read(data) @@ -347,7 +390,7 @@ class Handler attr_reader :content_type def initialize(io, km, callback, body_size_limit, decompression_size_limit, format_name, log, - cors_allow_origins, cors_allow_credentials, add_query_params) + cors_allow_origins, cors_allow_credentials, add_query_params, auth = nil) @io = io @km = km @callback = callback @@ -360,6 +403,8 @@ def initialize(io, km, callback, body_size_limit, decompression_size_limit, form @cors_allow_credentials = cors_allow_credentials @idle = 0 @add_query_params = add_query_params + @auth = auth + @authorization = nil @km.add(self) @remote_port, @remote_addr = io.remote_port, io.remote_addr @@ -399,6 +444,7 @@ def on_headers_complete(headers) @env = {} @content_type = "" @content_encoding = "" + @authorization = nil headers.each_pair {|k,v| @env["HTTP_#{k.tr('-','_').upcase}"] = v case k @@ -426,8 +472,22 @@ def on_headers_complete(headers) @access_control_request_method = v when /\AAccess-Control-Request-Headers\Z/i @access_control_request_headers = v + when /\AAuthorization\Z/i + @authorization = v.is_a?(Array) ? v.first : v end } + + if @auth + # Never let the credential reach a record built by add_http_headers. + @env.delete("HTTP_AUTHORIZATION") + + unless preflight_request? || authenticate + @log.warn "authentication failed", address: @remote_addr + send_response_and_close(RES_401_STATUS, {'WWW-Authenticate' => AUTH_CHALLENGE}, "") + return + end + end + if expect if expect == '100-continue'.freeze if !size || size < @body_size_limit @@ -442,6 +502,8 @@ def on_headers_complete(headers) end def on_body(chunk) + return if closing? + if @body.bytesize + chunk.bytesize > @body_size_limit unless closing? send_response_and_close("413 Request Entity Too Large", {}, "Too large") @@ -452,6 +514,8 @@ def on_body(chunk) end RES_200_STATUS = "200 OK".freeze + RES_401_STATUS = "401 Unauthorized".freeze + AUTH_CHALLENGE = 'Basic realm="fluentd"'.freeze RES_403_STATUS = "403 Forbidden".freeze # Azure App Service sends GET requests for health checking purpose. @@ -651,6 +715,23 @@ def include_cors_allow_origin def parse_query(query) query.nil? ? {} : Hash[URI.decode_www_form(query, Encoding::ASCII_8BIT)] end + + # Browsers never attach Authorization to a CORS preflight. + def preflight_request? + @parser.http_method == 'OPTIONS'.freeze + end + + def authenticate + scheme, credentials = @authorization.to_s.split(' ', 2) + return false unless scheme == 'Basic'.freeze + + username, password = credentials.to_s.unpack1('m').to_s.split(':', 2) + username_matched = OpenSSL.secure_compare(username.to_s, @auth.username) + password_matched = OpenSSL.secure_compare(password.to_s, @auth.password) + username_matched && password_matched + rescue ArgumentError + false + end end end end diff --git a/test/plugin/test_in_http.rb b/test/plugin/test_in_http.rb index 6ce2fb231a..d1ed61bfc7 100644 --- a/test/plugin/test_in_http.rb +++ b/test/plugin/test_in_http.rb @@ -53,6 +53,272 @@ def test_configure assert_equal false, d.instance.add_query_params end + def security_config(body, base = config) + base + %[ + + #{body} + + ] + end + + def test_configure_security_client_without_network + assert_raise(Fluent::ConfigError) do + create_driver(security_config("\n")) + end + end + + def test_configure_security_with_invalid_network + assert_raise(Fluent::ConfigError) do + create_driver(security_config("\n network 300.0.0.0/8\n")) + end + end + + def test_allowed_client_is_accepted + d = create_driver(security_config("\n network 127.0.0.0/8\n")) + time = event_time("2011-01-02 13:14:15 UTC") + time_i = time.to_i + + d.run(expect_records: 1) do + res = post("/tag1", {"json" => {"a" => 1}.to_json, "time" => time_i.to_s}) + assert_equal "200", res.code + end + + assert_equal [["tag1", time, {"a" => 1}]], d.events + end + + def test_disallowed_client_is_disconnected + d = create_driver(security_config("\n network 10.0.0.0/8\n")) + + d.run(expect_records: 0, timeout: 5) do + assert_raise(EOFError, Errno::ECONNRESET, Errno::EPIPE) do + post("/tag1", {"json" => {"a" => 1}.to_json}) + end + end + + assert_equal [], d.events + end + + def test_disallowed_client_cannot_spoof_with_x_forwarded_for + d = create_driver(security_config("\n network 10.0.0.0/8\n")) + + d.run(expect_records: 0, timeout: 5) do + assert_raise(EOFError, Errno::ECONNRESET, Errno::EPIPE) do + post("/tag1", {"json" => {"a" => 1}.to_json}, {"X-Forwarded-For" => "10.0.0.1"}) + end + end + + assert_equal [], d.events + end + + def test_allowed_client_matches_ipv4_and_ipv6_entries + d = create_driver(security_config(%[ + + network 10.0.0.0/8 + + + network 2001:db8::/32 + + ])) + instance = d.instance + + assert_true(instance.send(:allowed_client?, "10.1.2.3")) + assert_true(instance.send(:allowed_client?, "2001:db8::1")) + assert_false(instance.send(:allowed_client?, "11.1.2.3")) + assert_false(instance.send(:allowed_client?, "2001:db9::1")) + assert_false(instance.send(:allowed_client?, "not an address")) + end + + def test_security_without_client_accepts_any_peer + d = create_driver(security_config("")) + time = event_time("2011-01-02 13:14:15 UTC") + time_i = time.to_i + + d.run(expect_records: 1) do + res = post("/tag1", {"json" => {"a" => 1}.to_json, "time" => time_i.to_s}) + assert_equal "200", res.code + end + + assert_equal [["tag1", time, {"a" => 1}]], d.events + end + + def auth_config(base = config) + base + %[ + + method basic + username fluentd + password s3cret + + ] + end + + def basic_credentials(username = "fluentd", password = "s3cret") + "Basic " + ["#{username}:#{password}"].pack('m0') + end + + def test_configure_auth_without_username + assert_raise(Fluent::ConfigError) do + create_driver(config + %[ + + method basic + password s3cret + + ]) + end + end + + def read_http_response(sock) + status = sock.readline.split(' ')[1] + length = 0 + while (line = sock.readline) != "\r\n" + length = line.split(':', 2)[1].strip.to_i if /\AContent-Length:/i.match?(line) + end + sock.read(length) if length > 0 + status + end + + def test_basic_auth_accepts_valid_credentials + d = create_driver(auth_config) + time = event_time("2011-01-02 13:14:15 UTC") + time_i = time.to_i + + d.run(expect_records: 1) do + res = post("/tag1", {"json" => {"a" => 1}.to_json, "time" => time_i.to_s}, + {"Authorization" => basic_credentials}) + assert_equal "200", res.code + end + + assert_equal [["tag1", time, {"a" => 1}]], d.events + end + + def test_basic_auth_rejects_wrong_and_missing_credentials + d = create_driver(auth_config) + codes = [] + + d.end_if { codes.size == 4 } + d.run(expect_records: 0, timeout: 5) do + codes << post("/tag1", {"json" => "{}"}, {"Authorization" => basic_credentials("fluentd", "wrong")}).code + codes << post("/tag1", {"json" => "{}"}, {"Authorization" => basic_credentials("nobody", "s3cret")}).code + codes << post("/tag1", {"json" => "{}"}).code + codes << post("/tag1", {"json" => "{}"}, {"Authorization" => "Bearer s3cret"}).code + end + + assert_equal ["401", "401", "401", "401"], codes + assert_equal [], d.events + end + + class FakeIO + attr_reader :written + + def initialize + @written = +'' + end + + def remote_addr + "127.0.0.1" + end + + def remote_port + 12345 + end + + def write(data) + @written << data + end + + def close + end + end + + def build_handler(instance, io) + Fluent::Plugin::HttpInput::Handler.new( + io, + Fluent::Plugin::HttpInput::KeepaliveManager.new(10), + ->(_path, _params) { ["200 OK", {}, ""] }, + instance.body_size_limit, + instance.decompression_size_limit, + "default", + $log, + nil, + false, + false, + instance.auth + ) + end + + def feed_request(handler, authorization) + body = "x" * 65536 + header = +"POST /tag1 HTTP/1.1\r\nHost: 127.0.0.1\r\n" + header << "Authorization: #{authorization}\r\n" if authorization + header << "Content-Type: application/json\r\nContent-Length: #{body.bytesize}\r\n\r\n" + handler.on_read(header) + handler.on_read(body) + handler.instance_variable_get(:@body).bytesize + end + + def test_unauthorized_request_body_is_not_buffered + d = create_driver(auth_config) + + rejected_io = FakeIO.new + rejected = feed_request(build_handler(d.instance, rejected_io), nil) + + accepted = feed_request(build_handler(d.instance, FakeIO.new), basic_credentials) + + assert_true(rejected_io.written.start_with?("HTTP/1.1 401 Unauthorized")) + assert_equal(0, rejected) + assert_equal(65536, accepted) + end + + def test_authorization_header_is_kept_out_of_records + d = create_driver(auth_config(config + "add_http_headers true")) + time_i = event_time("2011-01-02 13:14:15 UTC").to_i + + d.run(expect_records: 1) do + res = post("/tag1", {"json" => {"a" => 1}.to_json, "time" => time_i.to_s}, + {"Authorization" => basic_credentials}) + assert_equal "200", res.code + end + + record = d.events[0][2] + assert_false(record.key?("HTTP_AUTHORIZATION")) + assert_equal(1, record["a"]) + end + + def test_cors_preflight_is_exempt_from_authentication + d = create_driver(auth_config(config + %[cors_allow_origins ["http://example.com"]])) + codes = [] + + d.end_if { codes.size == 1 } + d.run(expect_records: 0, timeout: 5) do + codes << options("/tag1", {}, {"Origin" => "http://example.com", + "Access-Control-Request-Method" => "POST"}).code + end + + assert_equal ["200"], codes + end + + def test_authorization_is_not_carried_over_a_keepalive_connection + d = create_driver(auth_config) + codes = [] + body = {"a" => 1}.to_json + + d.end_if { codes.size == 2 } + d.run(expect_records: 1, timeout: 5) do + TCPSocket.open("127.0.0.1", @port) do |sock| + sock.write("POST /tag1 HTTP/1.1\r\nHost: 127.0.0.1\r\n" \ + "Authorization: #{basic_credentials}\r\n" \ + "Content-Type: application/json\r\nContent-Length: #{body.bytesize}\r\n\r\n#{body}") + codes << read_http_response(sock) + + sock.write("POST /tag1 HTTP/1.1\r\nHost: 127.0.0.1\r\n" \ + "Content-Type: application/json\r\nContent-Length: #{body.bytesize}\r\n\r\n#{body}") + codes << read_http_response(sock) + end + end + + assert_equal ["200", "401"], codes + assert_equal 1, d.events.size + end + def test_time d = create_driver time = event_time("2011-01-02 13:14:15.123 UTC")