socket.rb

#
# 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 'socket'
require 'ipaddr'
require 'openssl'
if Fluent.windows?
  require 'certstore'
end

require 'fluent/tls'
require_relative 'socket_option'

module Fluent
  module PluginHelper
    module Socket
      # stop     : [-]
      # shutdown : [-]
      # close    : [-]
      # terminate: [-]

      include Fluent::PluginHelper::SocketOption

      attr_reader :_sockets # for tests

      # TODO: implement connection pool for specified host

      def socket_create(proto, host, port, **kwargs, &block)
        case proto
        when :tcp
          socket_create_tcp(host, port, **kwargs, &block)
        when :udp
          socket_create_udp(host, port, **kwargs, &block)
        when :tls
          socket_create_tls(host, port, **kwargs, &block)
        when :unix
          raise "not implemented yet"
        else
          raise ArgumentError, "invalid protocol: #{proto}"
        end
      end

      def socket_create_tcp(host, port, resolve_name: false, connect_timeout: nil, **kwargs, &block)
        sock = if connect_timeout
                 s = ::Socket.tcp(host, port, connect_timeout: connect_timeout)
                 s.autoclose = false # avoid GC triggered close
                 WrappedSocket::TCP.for_fd(s.fileno)
               else
                 WrappedSocket::TCP.new(host, port)
               end
        socket_option_set(sock, resolve_name: resolve_name, **kwargs)
        if block
          begin
            block.call(sock)
          ensure
            sock.close_write rescue nil
            sock.close rescue nil
          end
        else
          sock
        end
      end

      def socket_create_udp(host, port, resolve_name: false, connect: false, **kwargs, &block)
        family = IPAddr.new(IPSocket.getaddress(host)).ipv4? ? ::Socket::AF_INET : ::Socket::AF_INET6
        sock = WrappedSocket::UDP.new(family)
        socket_option_set(sock, resolve_name: resolve_name, **kwargs)
        sock.connect(host, port) if connect
        if block
          begin
            block.call(sock)
          ensure
            sock.close rescue nil
          end
        else
          sock
        end
      end

      def socket_create_tls(
          host, port,
          version: Fluent::TLS::DEFAULT_VERSION, min_version: nil, max_version: nil, ciphers: Fluent::TLS::CIPHERS_DEFAULT, insecure: false, verify_fqdn: true, fqdn: nil,
          enable_system_cert_store: true, allow_self_signed_cert: false, cert_paths: nil,
          cert_path: nil, private_key_path: nil, private_key_passphrase: nil,
          cert_thumbprint: nil, cert_logical_store_name: nil, cert_use_enterprise_store: true,
          connect_timeout: nil,
          **kwargs, &block)

        host_is_ipaddress = IPAddr.new(host) rescue false
        fqdn ||= host unless host_is_ipaddress

        context = OpenSSL::SSL::SSLContext.new

        if insecure
          log.trace "setting TLS verify_mode NONE"
          context.verify_mode = OpenSSL::SSL::VERIFY_NONE
        else
          cert_store = OpenSSL::X509::Store.new
          if allow_self_signed_cert && OpenSSL::X509.const_defined?('V_FLAG_CHECK_SS_SIGNATURE')
            cert_store.flags = OpenSSL::X509::V_FLAG_CHECK_SS_SIGNATURE
          end
          begin
            if enable_system_cert_store
              if Fluent.windows? && cert_logical_store_name
                log.trace "loading Windows system certificate store"
                loader = Certstore::OpenSSL::Loader.new(log, cert_store, cert_logical_store_name,
                                                        enterprise: cert_use_enterprise_store)
                loader.load_cert_store
                cert_store = loader.cert_store
                context.cert = loader.get_certificate(cert_thumbprint) if cert_thumbprint
              end
              log.trace "loading system default certificate store"
              cert_store.set_default_paths
            end
          rescue OpenSSL::X509::StoreError
            log.warn "failed to load system default certificate store", error: e
          end
          if cert_paths
            if cert_paths.respond_to?(:each)
              cert_paths.each do |cert_path|
                log.trace "adding CA cert", path: cert_path
                cert_store.add_file(cert_path)
              end
            else
              cert_path = cert_paths
              log.trace "adding CA cert", path: cert_path
              cert_store.add_file(cert_path)
            end
          end

          log.trace "setting TLS context", mode: "peer", ciphers: ciphers
          context.set_params({})
          context.ciphers = ciphers
          context.verify_mode = OpenSSL::SSL::VERIFY_PEER
          context.cert_store = cert_store
          context.verify_hostname = verify_fqdn && fqdn
          context.key = OpenSSL::PKey::read(File.read(private_key_path), private_key_passphrase) if private_key_path

          if cert_path
            certs = socket_certificates_from_file(cert_path)
            context.cert = certs.shift
            unless certs.empty?
              context.extra_chain_cert = certs
            end
          end
        end
        Fluent::TLS.set_version_to_context(context, version, min_version, max_version)

        tcpsock = socket_create_tcp(host, port, connect_timeout: connect_timeout, **kwargs)
        sock = WrappedSocket::TLS.new(tcpsock, context)
        sock.sync_close = true
        sock.hostname = fqdn if verify_fqdn && fqdn && sock.respond_to?(:hostname=)

        log.trace "entering TLS handshake"
        if connect_timeout
          begin
            Timeout.timeout(connect_timeout) { sock.connect }
          rescue Timeout::Error
            log.warn "timeout while connecting tls session", host: host
            sock.close rescue nil
            raise
          end
        else
          sock.connect
        end

        begin
          if verify_fqdn
            log.trace "checking peer's certificate", subject: sock.peer_cert.subject
            sock.post_connection_check(fqdn)
            verify = sock.verify_result
            if verify != OpenSSL::X509::V_OK
              err_name = Socket.tls_verify_result_name(verify)
              log.warn "BUG: failed to verify certification while connecting (but not raised, why?)", host: host, fqdn: fqdn, error: err_name
              raise RuntimeError, "BUG: failed to verify certification and to handle it correctly while connecting host #{host} as #{fqdn}"
            end
          end
        rescue OpenSSL::SSL::SSLError => e
          log.warn "failed to verify certification while connecting tls session", host: host, fqdn: fqdn, error: e
          raise
        end

        if block
          begin
            block.call(sock)
          ensure
            sock.close rescue nil
          end
        else
          sock
        end
      end

      def socket_certificates_from_file(path)
        data = File.read(path)
        pattern = Regexp.compile('-+BEGIN CERTIFICATE-+\r?\n(?:[^-]*\r?\n)+-+END CERTIFICATE-+\r?\n?', Regexp::MULTILINE)
        list = []
        data.scan(pattern) { |match| list << OpenSSL::X509::Certificate.new(match) }
        if list.length == 0
          raise Fluent::ConfigError, "cert_path does not contain a valid certificate"
        end
        list
      end

      def self.tls_verify_result_name(code)
        case code
        when OpenSSL::X509::V_OK then 'V_OK'
        when OpenSSL::X509::V_ERR_AKID_SKID_MISMATCH then 'V_ERR_AKID_SKID_MISMATCH'
        when OpenSSL::X509::V_ERR_APPLICATION_VERIFICATION then 'V_ERR_APPLICATION_VERIFICATION'
        when OpenSSL::X509::V_ERR_CERT_CHAIN_TOO_LONG then 'V_ERR_CERT_CHAIN_TOO_LONG'
        when OpenSSL::X509::V_ERR_CERT_HAS_EXPIRED then 'V_ERR_CERT_HAS_EXPIRED'
        when OpenSSL::X509::V_ERR_CERT_NOT_YET_VALID then 'V_ERR_CERT_NOT_YET_VALID'
        when OpenSSL::X509::V_ERR_CERT_REJECTED then 'V_ERR_CERT_REJECTED'
        when OpenSSL::X509::V_ERR_CERT_REVOKED then 'V_ERR_CERT_REVOKED'
        when OpenSSL::X509::V_ERR_CERT_SIGNATURE_FAILURE then 'V_ERR_CERT_SIGNATURE_FAILURE'
        when OpenSSL::X509::V_ERR_CERT_UNTRUSTED then 'V_ERR_CERT_UNTRUSTED'
        when OpenSSL::X509::V_ERR_CRL_HAS_EXPIRED then 'V_ERR_CRL_HAS_EXPIRED'
        when OpenSSL::X509::V_ERR_CRL_NOT_YET_VALID then 'V_ERR_CRL_NOT_YET_VALID'
        when OpenSSL::X509::V_ERR_CRL_SIGNATURE_FAILURE then 'V_ERR_CRL_SIGNATURE_FAILURE'
        when OpenSSL::X509::V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT then 'V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT'
        when OpenSSL::X509::V_ERR_ERROR_IN_CERT_NOT_AFTER_FIELD then 'V_ERR_ERROR_IN_CERT_NOT_AFTER_FIELD'
        when OpenSSL::X509::V_ERR_ERROR_IN_CERT_NOT_BEFORE_FIELD then 'V_ERR_ERROR_IN_CERT_NOT_BEFORE_FIELD'
        when OpenSSL::X509::V_ERR_ERROR_IN_CRL_LAST_UPDATE_FIELD then 'V_ERR_ERROR_IN_CRL_LAST_UPDATE_FIELD'
        when OpenSSL::X509::V_ERR_ERROR_IN_CRL_NEXT_UPDATE_FIELD then 'V_ERR_ERROR_IN_CRL_NEXT_UPDATE_FIELD'
        when OpenSSL::X509::V_ERR_INVALID_CA then 'V_ERR_INVALID_CA'
        when OpenSSL::X509::V_ERR_INVALID_PURPOSE then 'V_ERR_INVALID_PURPOSE'
        when OpenSSL::X509::V_ERR_KEYUSAGE_NO_CERTSIGN then 'V_ERR_KEYUSAGE_NO_CERTSIGN'
        when OpenSSL::X509::V_ERR_OUT_OF_MEM then 'V_ERR_OUT_OF_MEM'
        when OpenSSL::X509::V_ERR_PATH_LENGTH_EXCEEDED then 'V_ERR_PATH_LENGTH_EXCEEDED'
        when OpenSSL::X509::V_ERR_SELF_SIGNED_CERT_IN_CHAIN then 'V_ERR_SELF_SIGNED_CERT_IN_CHAIN'
        when OpenSSL::X509::V_ERR_SUBJECT_ISSUER_MISMATCH then 'V_ERR_SUBJECT_ISSUER_MISMATCH'
        when OpenSSL::X509::V_ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY then 'V_ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY'
        when OpenSSL::X509::V_ERR_UNABLE_TO_DECRYPT_CERT_SIGNATURE then 'V_ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY'
        when OpenSSL::X509::V_ERR_UNABLE_TO_DECRYPT_CRL_SIGNATURE then 'V_ERR_UNABLE_TO_DECRYPT_CRL_SIGNATURE'
        when OpenSSL::X509::V_ERR_UNABLE_TO_GET_CRL then 'V_ERR_UNABLE_TO_GET_CRL'
        when OpenSSL::X509::V_ERR_UNABLE_TO_GET_ISSUER_CERT then 'V_ERR_UNABLE_TO_GET_ISSUER_CERT'
        when OpenSSL::X509::V_ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY then 'V_ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY'
        when OpenSSL::X509::V_ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE then 'V_ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE'
        end
      end

      # socket_create_socks ?

      module WrappedSocket
        class TCP < ::TCPSocket
          def remote_addr; peeraddr[3]; end
          def remote_host; peeraddr[2]; end
          def remote_port; peeraddr[1]; end
        end
        class UDP < ::UDPSocket
          def remote_addr; peeraddr[3]; end
          def remote_host; peeraddr[2]; end
          def remote_port; peeraddr[1]; end
        end
        class TLS < OpenSSL::SSL::SSLSocket
          def remote_addr; peeraddr[3]; end
          def remote_host; peeraddr[2]; end
          def remote_port; peeraddr[1]; end
        end
      end

      def initialize
        super
        # @_sockets = [] # for keepalived sockets / connection pool
      end

      # def close
      #   @_sockets.each do |sock|
      #     sock.close
      #   end
      #   super
      # end
    end
  end
end