HTTP server helper supports https

Gemfile

@@ -2,8 +2,6 @@ source 'https://rubygems.org/'
 
 gemspec
 
-gem 'async-http', '~> 0.42'
-
 local_gemfile = File.join(File.dirname(__FILE__), "Gemfile.local")
 if File.exist?(local_gemfile)
   puts "Loading Gemfile.local ..." if $DEBUG # `ruby -d` or `bundle -v`

fluentd.gemspec

@@ -49,4 +49,5 @@ Gem::Specification.new do |gem|
   gem.add_development_dependency("test-unit-rr", ["~> 1.0"])
   gem.add_development_dependency("oj", [">= 2.14", "< 4"])
   gem.add_development_dependency("ext_monitor", [">= 0.1.1", "< 0.2"])
+  gem.add_development_dependency("async-http")
 end

lib/fluent/plugin/in_monitor_agent.rb

@@ -210,7 +210,7 @@ def start
 
       log.debug "listening monitoring http server on http://#{@bind}:#{@port}/api/plugins for worker#{fluentd_worker_id}"
       api_handler = APIHandler.new(self)
-      create_http_server(:in_monitor_http_server_helper, addr: @bind, port: @port, logger: log, default_app: NotFoundJson) do |serv|
+      http_server_create_http_server(:in_monitor_http_server_helper, addr: @bind, port: @port, logger: log, default_app: NotFoundJson) do |serv|
         serv.get('/api/plugins') { |req| api_handler.plugins_ltsv(req) }
         serv.get('/api/plugins.json') { |req| api_handler.plugins_json(req) }
         serv.get('/api/config') { |req| api_handler.config_ltsv(req) }

lib/fluent/plugin_helper/cert_option.rb

@@ -17,7 +17,7 @@
 require 'openssl'
 require 'socket'
 
-# this module is only for Socket/Server plugin helpers
+# this module is only for Socket/Server/HttpServer plugin helpers
 module Fluent
   module PluginHelper
     module CertOption

lib/fluent/plugin_helper/http_server.rb

@@ -23,27 +23,77 @@
 end
 
 require 'fluent/plugin_helper/thread'
+require 'fluent/plugin_helper/server' # For Server::ServerTransportParams
+require 'fluent/plugin_helper/http_server/ssl_context_builder'
 
 module Fluent
   module PluginHelper
     module HttpServer
       include Fluent::PluginHelper::Thread
+      include Fluent::Configurable
+
       # stop     : stop http server and mark callback thread as stopped
       # shutdown : [-]
       # close    : correct stopped threads
       # terminate: kill thread
 
+      def self.included(mod)
+        mod.include Fluent::PluginHelper::Server::ServerTransportParams
+      end
+
+      def initialize(*)
+        super
+        @_http_server = nil
+      end
+
+      def create_http_server(title, addr:, port:, logger:, default_app: nil, proto: nil, tls_opts: nil, &block)
+        logger.warn('this method is deprecated. Use #http_server_create_http_server instead')
+        http_server_create_http_server(title, addr: addr, port: port, logger: logger, default_app: default_app, proto: proto, tls_opts: tls_opts, &block)
+      end
+
       # @param title [Symbol] the thread name. this value should be unique.
       # @param addr [String] Listen address
       # @param port [String] Listen port
       # @param logger [Logger] logger used in this server
       # @param default_app [Object] This method must have #call.
-      def create_http_server(title, addr:, port:, logger:, default_app: nil)
+      # @param proto [Symbol] :tls or :tcp
+      # @param tls_opts [Hash] options for TLS.
+      def http_server_create_http_server(title, addr:, port:, logger:, default_app: nil, proto: nil, tls_opts: nil, &block)
         unless block_given?
           raise ArgumentError, 'BUG: callback not specified'
         end
 
-        @_http_server = HttpServer::Server.new(addr: addr, port: port, logger: logger, default_app: default_app) do |serv|
+        if proto == :tls || (@transport_config && @transport_config.protocol == :tls)
+          http_server_create_https_server(title, addr: addr, port: port, logger: logger, default_app: default_app, tls_opts: tls_opts, &block)
+        else
+          @_http_server = HttpServer::Server.new(addr: addr, port: port, logger: logger, default_app: default_app) do |serv|
+            yield(serv)
+          end
+
+          _block_until_http_server_start do |notify|
+            thread_create(title) do
+              @_http_server.start(notify)
+            end
+          end
+        end
+      end
+
+      # @param title [Symbol] the thread name. this value should be unique.
+      # @param addr [String] Listen address
+      # @param port [String] Listen port
+      # @param logger [Logger] logger used in this server
+      # @param default_app [Object] This method must have #call.
+      # @param tls_opts [Hash] options for TLS.
+      def http_server_create_https_server(title, addr:, port:, logger:, default_app: nil, tls_opts: nil)
+        topt =
+          if tls_opts
+            _http_server_overwrite_config(@transport_config, tls_opts)
+          else
+            @transport_config
+          end
+        ctx = Fluent::PluginHelper::HttpServer::SSLContextBuilder.new($log).build(topt)
+
+        @_http_server = HttpServer::Server.new(addr: addr, port: port, logger: logger, default_app: default_app, tls_context: ctx) do |serv|
           yield(serv)
         end
 
@@ -64,6 +114,16 @@ def stop
 
       private
 
+      def _http_server_overwrite_config(config, opts)
+        conf = config
+        Fluent::PluginHelper::Server::SERVER_TRANSPORT_PARAMS.map(&:to_s).each do |param|
+          if opts.key?(param)
+            conf[param] = opts[param]
+          end
+        end
+        conf
+      end
+
       # To block until server is ready to listen
       def _block_until_http_server_start
         que = Queue.new

lib/fluent/plugin_helper/http_server/compat/server.rb

@@ -16,6 +16,7 @@
 
 require 'fluent/plugin_helper/http_server/methods'
 require 'fluent/plugin_helper/http_server/compat/webrick_handler'
+require 'fluent/plugin_helper/http_server/compat/ssl_context_extractor'
 
 module Fluent
   module PluginHelper
@@ -24,16 +25,26 @@ module Compat
         class Server
           # @param logger [Logger]
           # @param default_app [Object] ignored option. only for compat
-          def initialize(addr:, port:, logger:, default_app: nil)
+          # @param tls_context [OpenSSL::SSL::SSLContext]
+          def initialize(addr:, port:, logger:, default_app: nil, tls_context: nil)
             @addr = addr
             @port = port
             @logger = logger
-            @server = WEBrick::HTTPServer.new(
+
+            config = {
               BindAddress: @addr,
               Port: @port,
               Logger: WEBrick::Log.new(STDERR, WEBrick::Log::FATAL),
               AccessLog: [],
-            )
+            }
+            if tls_context
+              require 'webrick/https'
+              @logger.warn('Webrick ignores given TLS version')
+              tls_opt = Fluent::PluginHelper::HttpServer::Compat::SSLContextExtractor.extract(tls_context)
+              config = tls_opt.merge(**config)
+            end
+
+            @server = WEBrick::HTTPServer.new(config)
 
             # @example ["/example.json", :get, handler object]
             @methods = []

lib/fluent/plugin_helper/http_server/compat/ssl_context_extractor.rb

@@ -0,0 +1,52 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+module Fluent
+  module PluginHelper
+    module HttpServer
+      module Compat
+        # This class converts OpenSSL::SSL::SSLContext to Webrick SSL Config because webrick does not have interface to pass OpenSSL::SSL::SSLContext directory
+        # https://github.com/ruby/webrick/blob/v1.6.0/lib/webrick/ssl.rb#L67-L88
+        class SSLContextExtractor
+
+          #
+          # memo: https://github.com/ruby/webrick/blob/v1.6.0/lib/webrick/ssl.rb#L180-L205
+          # @param ctx [OpenSSL::SSL::SSLContext]
+          def self.extract(ctx)
+            {
+              SSLEnable: true,
+              SSLPrivateKey: ctx.key,
+              SSLCertificate: ctx.cert,
+              SSLClientCA: ctx.client_ca,
+              SSLExtraChainCert: ctx.extra_chain_cert,
+              SSLCACertificateFile: ctx.ca_file,
+              SSLCACertificatePath: ctx.ca_path,
+              SSLCertificateStore: ctx.cert_store,
+              SSLTmpDhCallback: ctx.tmp_dh_callback,
+              SSLVerifyClient: ctx.verify_mode,
+              SSLVerifyDepth: ctx.verify_depth,
+              SSLVerifyCallback: ctx.verify_callback,
+              SSLServerNameCallback: ctx.servername_cb,
+              SSLTimeout: ctx.timeout,
+              SSLOptions: ctx.options,
+              SSLCiphers: ctx.ciphers,
+            }
+          end
+        end
+      end
+    end
+  end
+end

lib/fluent/plugin_helper/http_server/server.rb

@@ -26,20 +26,26 @@ module Fluent
   module PluginHelper
     module HttpServer
       class Server
+        # @param logger [Logger]
         # @param default_app [Object] This method must have #call.
-        def initialize(addr:, port:, logger:, default_app: nil)
+        # @param tls_context [OpenSSL::SSL::SSLContext]
+        def initialize(addr:, port:, logger:, default_app: nil, tls_context: nil)
           @addr = addr
           @port = port
           @logger = logger
 
-          # TODO: support https and http2
-          @uri = URI("http://#{@addr}:#{@port}").to_s
+          # TODO: support http2
+          scheme = tls_context ? 'https' : 'http'
+          @uri = URI("#{scheme}://#{@addr}:#{@port}").to_s
           @router = Router.new(default_app)
-          @reactor = Async::Reactor.new
-          @server = Async::HTTP::Server.new(
-            App.new(@router, @logger),
-            Async::HTTP::Endpoint.parse(@uri)
-          )
+          @reactor = Async::Reactor.new(nil, logger: @logger)
+
+          opts = if tls_context
+                   { ssl_context: tls_context }
+                 else
+                   {}
+                 end
+          @server = Async::HTTP::Server.new(App.new(@router, @logger), Async::HTTP::Endpoint.parse(@uri, **opts))
 
           if block_given?
             yield(self)

lib/fluent/plugin_helper/http_server/ssl_context_builder.rb

@@ -0,0 +1,41 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'fluent/plugin_helper/cert_option'
+
+module Fluent
+  module PluginHelper
+    module HttpServer
+      # In order not to expose CertOption's methods unnecessary
+      class SSLContextBuilder
+        include Fluent::PluginHelper::CertOption
+
+        def initialize(log)
+          @log = log
+        end
+
+        # @param config [Fluent::Config::Section] @transport_config
+        def build(config)
+          cert_option_create_context(config.version, config.insecure, config.ciphers, config)
+        end
+
+        private
+
+        attr_reader :log
+      end
+    end
+  end
+end

test/plugin_helper/data/cert/generate_cert.rb

@@ -0,0 +1,87 @@
+require 'fluent/plugin_helper/cert_option'
+require 'fileutils'
+
+module CertUtil
+  extend Fluent::PluginHelper::CertOption
+end
+
+WITHOUT_CA_DIR = './without_ca'.freeze
+WITH_CA_DIR = './with_ca'.freeze
+
+CA_OPTION = {
+  private_key_length: 2048,
+  country: 'US',
+  state: 'CA',
+  locality: 'Mountain View',
+  common_name: 'ca.testing.fluentd.org',
+  expiration: 30 * 86400,
+  digest: :sha256,
+}
+
+SERVER_OPTION = {
+  private_key_length: 2048,
+  country: 'US',
+  state: 'CA',
+  locality: 'Mountain View',
+  common_name: 'server.testing.fluentd.org',
+  expiration: 30 * 86400,
+  digest: :sha256,
+}
+
+def write_cert_and_key(cert_path, cert, key_path, key, passphrase)
+  File.open(cert_path, 'w') { |f| f.write(cert.to_pem) }
+
+  # Write the secret key (raw or encrypted by AES256) in PEM format
+  key_str = passphrase ? key.export(OpenSSL::Cipher.new('AES-256-CBC'), passphrase) : key.export
+  File.open(key_path, 'w') { |f| f.write(key_str) }
+  File.chmod(0o600, cert_path, key_path)
+end
+
+def create_server_pair_signed_by_self(cert_path, private_key_path, passphrase)
+  cert, key, _ = CertUtil.cert_option_generate_server_pair_self_signed(SERVER_OPTION)
+  write_cert_and_key(cert_path, cert, private_key_path, key, passphrase)
+  cert
+end
+
+def create_ca_pair_signed_by_self(cert_path, private_key_path, passphrase)
+  cert, key, _ = CertUtil.cert_option_generate_ca_pair_self_signed(CA_OPTION)
+  write_cert_and_key(cert_path, cert, private_key_path, key, passphrase)
+  cert
+end
+
+def create_server_pair_signed_by_ca(ca_cert_path, ca_key_path, ca_key_passphrase, cert_path, private_key_path, passphrase)
+  cert, key, _ = CertUtil.cert_option_generate_server_pair_by_ca(ca_cert_path, ca_key_path, ca_key_passphrase, SERVER_OPTION)
+  write_cert_and_key(cert_path, cert, private_key_path, key, passphrase)
+  cert
+end
+
+def create_without_ca
+  FileUtils.mkdir_p(WITHOUT_CA_DIR)
+  cert_path = File.join(WITHOUT_CA_DIR, 'cert.pem')
+  cert_key_path = File.join(WITHOUT_CA_DIR, 'cert-key.pem')
+  cert_pass_path = File.join(WITHOUT_CA_DIR, 'cert-pass.pem')
+  cert_key_pass_path = File.join(WITHOUT_CA_DIR, 'cert-key-pass.pem')
+
+  create_server_pair_signed_by_self(cert_path, cert_key_path, nil)
+  create_server_pair_signed_by_self(cert_pass_path, cert_key_pass_path, 'apple') # with passphrase
+end
+
+def create_with_ca
+  FileUtils.mkdir_p(WITH_CA_DIR)
+  cert_path = File.join(WITH_CA_DIR, 'cert.pem')
+  cert_key_path = File.join(WITH_CA_DIR, 'cert-key.pem')
+  ca_cert_path = File.join(WITH_CA_DIR, 'ca-cert.pem')
+  ca_key_path = File.join(WITH_CA_DIR, 'ca-cert-key.pem')
+  create_ca_pair_signed_by_self(ca_cert_path, ca_key_path, nil)
+  create_server_pair_signed_by_ca(ca_cert_path, ca_key_path, nil, cert_path, cert_key_path, nil)
+
+  cert_pass_path = File.join(WITH_CA_DIR, 'cert-pass.pem')
+  cert_key_pass_path = File.join(WITH_CA_DIR, 'cert-key-pass.pem')
+  ca_cert_pass_path = File.join(WITH_CA_DIR, 'ca-cert-pass.pem')
+  ca_key_pass_path = File.join(WITH_CA_DIR, 'ca-cert-key-pass.pem')
+  create_ca_pair_signed_by_self(ca_cert_pass_path, ca_key_pass_path, 'orange') # with passphrase
+  create_server_pair_signed_by_ca(ca_cert_pass_path, ca_key_pass_path, 'orange', cert_pass_path, cert_key_pass_path, 'apple')
+end
+
+create_without_ca
+create_with_ca