Add support for using the system's default CA cert store

Author: bradurani

README.md

@@ -888,6 +888,21 @@ kafka = Kafka.new(
 
 Without passing the CA certificate to the client it would be impossible to protect against [man-in-the-middle attacks](https://en.wikipedia.org/wiki/Man-in-the-middle_attack).
 
+##### Using your system's CA cert store
+
+If you want to use the CA certs from your system's default certificate store, you
+can use:
+
+```ruby
+kafka = Kafka.new(
+  use_ssl_ca_default_store: true
+  # ...
+)
+```
+
+This configures the store to look up CA certificates from the system default certificate store on an as needed basis. The location of the store can usually be determined by: 
+`OpenSSL::X509::DEFAULT_CERT_FILE`
+
 ##### Client Authentication
 
 In order to authenticate the client to the cluster, you need to pass in a certificate and key created for the client and trusted by the brokers.

lib/kafka/client.rb

@@ -60,12 +60,12 @@ def initialize(seed_brokers:, client_id: "ruby-kafka", logger: nil, connect_time
                    ssl_ca_cert_file_path: nil, ssl_ca_cert: nil, ssl_client_cert: nil, ssl_client_cert_key: nil,
                    sasl_gssapi_principal: nil, sasl_gssapi_keytab: nil,
                    sasl_plain_authzid: '', sasl_plain_username: nil, sasl_plain_password: nil,
-                   sasl_scram_username: nil, sasl_scram_password: nil, sasl_scram_mechanism: nil)
+                   sasl_scram_username: nil, sasl_scram_password: nil, sasl_scram_mechanism: nil, use_ssl_ca_default_store: false)
       @logger = logger || Logger.new(nil)
       @instrumenter = Instrumenter.new(client_id: client_id)
       @seed_brokers = normalize_seed_brokers(seed_brokers)
 
-      ssl_context = build_ssl_context(ssl_ca_cert_file_path, ssl_ca_cert, ssl_client_cert, ssl_client_cert_key)
+      ssl_context = build_ssl_context(ssl_ca_cert_file_path, ssl_ca_cert, ssl_client_cert, ssl_client_cert_key, use_ssl_ca_default_store)
 
       sasl_authenticator = SaslAuthenticator.new(
         sasl_gssapi_principal: sasl_gssapi_principal,
@@ -542,8 +542,8 @@ def initialize_cluster
       )
     end
 
-    def build_ssl_context(ca_cert_file_path, ca_cert, client_cert, client_cert_key)
-      return nil unless ca_cert_file_path || ca_cert || client_cert || client_cert_key
+    def build_ssl_context(ca_cert_file_path, ca_cert, client_cert, client_cert_key, use_ssl_ca_default_store)
+      return nil unless ca_cert_file_path || ca_cert || client_cert || client_cert_key || use_ssl_ca_default_store
 
       ssl_context = OpenSSL::SSL::SSLContext.new
 
@@ -558,14 +558,17 @@ def build_ssl_context(ca_cert_file_path, ca_cert, client_cert, client_cert_key)
         raise ArgumentError, "Kafka client initialized with `ssl_client_cert_key`, but no `ssl_client_cert`. Please provide both."
       end
 
-      if ca_cert || ca_cert_file_path
+      if ca_cert || ca_cert_file_path || use_ssl_ca_default_store
         store = OpenSSL::X509::Store.new
         Array(ca_cert).each do |cert|
           store.add_cert(OpenSSL::X509::Certificate.new(cert))
         end
         if ca_cert_file_path
           store.add_file(ca_cert_file_path)
         end
+        if use_ssl_ca_default_store
+          store.set_default_paths
+        end
         ssl_context.cert_store = store
       end