generate_certs.sh

#!/bin/bash

set -e
set -v

OUTPUT_DIR=/Users/jturkel/dev/confluent-2.0.1/ssl
rm -rf $OUTPUT_DIR
mkdir $OUTPUT_DIR

# Generate a certificate authority
openssl req -new -newkey rsa:2048 -x509 -keyout $OUTPUT_DIR/ca-key -out $OUTPUT_DIR/ca-cert -days 365 -nodes -subj "/C=US/ST=MA/L=Boston/O=Salsify/OU=Product/CN=Certificates R Us"

################ Setup the server keystore ########################

# Create a new key pair
keytool -genkeypair -keystore $OUTPUT_DIR/server.jks -storepass password -keypass password -alias localhost -validity 365 -keyalg RSA -keysize 2048 -dname "cn=kafka.server.local, ou=Product, o=Salsify, c=US"

# Import the CA's certificate into the keystore
keytool -keystore $OUTPUT_DIR/server.jks -storepass password -alias CARoot -import -noprompt -file $OUTPUT_DIR/ca-cert

# Create a signing request
keytool -keystore $OUTPUT_DIR/server.jks -storepass password -alias localhost -certreq -file $OUTPUT_DIR/server.csr

# Use the CA to sign the certificate
openssl x509 -req -CA $OUTPUT_DIR/ca-cert -CAkey $OUTPUT_DIR/ca-key -in $OUTPUT_DIR/server.csr -out $OUTPUT_DIR/server.crt -days 365 -CAcreateserial 

# Import the signed certificate
keytool -keystore $OUTPUT_DIR/server.jks -storepass password -alias localhost -import -noprompt -file $OUTPUT_DIR/server.crt


################ Setup the client keystore ########################

# Create a new key pair
keytool -genkeypair -keystore $OUTPUT_DIR/client.jks -storepass password -keypass password -alias localhost -validity 365 -keyalg RSA -keysize 2048 -dname "cn=kafka.client.local, ou=Product, o=Salsify, c=US"

# Import the CA's certificate into the keystore
keytool -keystore $OUTPUT_DIR/client.jks -storepass password -alias CARoot -import -noprompt -file $OUTPUT_DIR/ca-cert

# Create a signing request
keytool -keystore $OUTPUT_DIR/client.jks -storepass password -alias localhost -certreq -file $OUTPUT_DIR/client.csr

# Use the CA to sign the certificate
openssl x509 -req -CA $OUTPUT_DIR/ca-cert -CAkey $OUTPUT_DIR/ca-key -in $OUTPUT_DIR/client.csr -out $OUTPUT_DIR/client.crt -days 365 -CAcreateserial 

# Import the signed certificate
keytool -keystore $OUTPUT_DIR/client.jks -storepass password -alias localhost -import -noprompt -file $OUTPUT_DIR/client.crt

# Convert the client keystore to pkcs12 then to pem
keytool -importkeystore -srckeystore $OUTPUT_DIR/client.jks -destkeystore $OUTPUT_DIR/client.p12 -srcstoretype jks -deststoretype pkcs12 -srcstorepass password -deststorepass password -noprompt -srcalias localhost
openssl pkcs12 -in $OUTPUT_DIR/client.p12 -out $OUTPUT_DIR/client.pem -nodes -passin pass:password

socket_with_timeout_patch.rb

Kafka::SocketWithTimeout.class_eval do
  def initialize(host, port, connect_timeout: nil, timeout: nil)
    @timeout = timeout
    
    ssl_context = OpenSSL::SSL::SSLContext.new
    ssl_context.ca_file = '/Users/jturkel/dev/confluent-2.0.1/ssl/ca-cert'
    ssl_context.verify_mode = OpenSSL::SSL::VERIFY_PEER
    ssl_context.cert = OpenSSL::X509::Certificate.new(File.read('/Users/jturkel/dev/confluent-2.0.1/ssl/client.crt'))
    ssl_context.key = OpenSSL::PKey::RSA.new(File.read('/Users/jturkel/dev/confluent-2.0.1/ssl/client.pem'))
    
    plain_socket = TCPSocket.new(host, port)
    @socket = OpenSSL::SSL::SSLSocket.new(plain_socket, ssl_context)
    @socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)

    begin
      # Initiate the socket connection in the background. If it doesn't fail
      # immediately it will raise an IO::WaitWritable (Errno::EINPROGRESS)
      # indicating the connection is in progress.
      @socket.connect_nonblock
    rescue IO::WaitReadable
      # IO.select will block until the socket is writable or the timeout
      # is exceeded, whichever comes first.
      unless IO.select(nil, [@socket], nil, connect_timeout)
        # IO.select returns nil when the socket is not ready before timeout
        # seconds have elapsed
        @socket.close
        raise Errno::ETIMEDOUT
      end

      begin
        # Verify there is now a good connection.
        @socket.connect_nonblock
      rescue IO::WaitReadable
        # The socket is connected, we're good!
      end
    end
  end

  def set_encoding(encoding)
    # TODO: OpenSSL::SSL::SSLSocket doesn't have a set_encoding method
    # @socket.set_encoding(encoding)
  end
end