Require and verify client certificate if proxy-listener-ca-chain-cert-file is provided

Author: everesio

README.md

@@ -105,7 +105,7 @@ See:
           --kafka-write-timeout duration               How long to wait for a transmit (default 30s)
           --log-format string                          Log format text or json (default "text")
           --log-level string                           Log level debug, info, warning, error, fatal or panic (default "info")
-          --proxy-listener-ca-chain-cert-file string   PEM encoded CA's certificate file
+          --proxy-listener-ca-chain-cert-file string   PEM encoded CA's certificate file. If provided, client certificate is required and verified
           --proxy-listener-cert-file string            PEM encoded file with server certificate
           --proxy-listener-keep-alive duration         Keep alive period for an active network connection. If zero, keep-alives are disabled (default 1m0s)
           --proxy-listener-key-file string             PEM encoded file with private key for the server certificate
@@ -316,7 +316,7 @@ spec:
   1. additional handshake - protocol: magic, method, data
   2. google-id method
 * [X] Registry for built-in plugins
-* [ ] Client cert check
+* [X] Client cert check
 * [ ] TLS server parameters like CipherSuites etc. - see ory/graceful/blob/master/http_defaults.go
 * [ ] Performance tests and tuning
 * [ ] Socket buffer sizing e.g. SO_RCVBUF = 32768, SO_SNDBUF = 131072

cmd/kafka-proxy/server.go

@@ -81,7 +81,7 @@ func init() {
 	Server.Flags().StringVar(&c.Proxy.TLS.ListenerCertFile, "proxy-listener-cert-file", "", "PEM encoded file with server certificate")
 	Server.Flags().StringVar(&c.Proxy.TLS.ListenerKeyFile, "proxy-listener-key-file", "", "PEM encoded file with private key for the server certificate")
 	Server.Flags().StringVar(&c.Proxy.TLS.ListenerKeyPassword, "proxy-listener-key-password", "", "Password to decrypt rsa private key")
-	Server.Flags().StringVar(&c.Proxy.TLS.CAChainCertFile, "proxy-listener-ca-chain-cert-file", "", "PEM encoded CA's certificate file")
+	Server.Flags().StringVar(&c.Proxy.TLS.CAChainCertFile, "proxy-listener-ca-chain-cert-file", "", "PEM encoded CA's certificate file. If provided, client certificate is required and verified")
 
 	// local authentication plugin
 	Server.Flags().BoolVar(&c.Auth.Local.Enable, "auth-local-enable", false, "Enable local SASL/PLAIN authentication performed by listener - SASL handshake will not be passed to kafka brokers")

proxy/auth_test.go

@@ -7,15 +7,14 @@ import (
 	"fmt"
 	"github.com/grepplabs/kafka-proxy/pkg/apis"
 	"github.com/stretchr/testify/assert"
-	"net"
 	"testing"
 	"time"
 )
 
 func TestAuthHandshake(t *testing.T) {
 	a := assert.New(t)
 
-	magic, err := Uint64()
+	magic, err := RandomUint64()
 	a.Nil(err)
 
 	fmt.Println(magic)
@@ -83,45 +82,7 @@ func (p *testTokenInfo) VerifyToken(ctx context.Context, request apis.VerifyRequ
 	return apis.VerifyResponse{Success: false}, p.err
 }
 
-func makePipe() (c1, c2 net.Conn, stop func(), err error) {
-	ln, err := net.Listen("tcp4", "127.0.0.1:0")
-	if err != nil {
-		return nil, nil, nil, err
-	}
-
-	// Start a connection between two endpoints.
-	var err1, err2 error
-	done := make(chan bool)
-	go func() {
-		c2, err2 = ln.Accept()
-		close(done)
-	}()
-	c1, err1 = net.Dial(ln.Addr().Network(), ln.Addr().String())
-	<-done
-
-	stop = func() {
-		if err1 == nil {
-			c1.Close()
-		}
-		if err2 == nil {
-			c2.Close()
-		}
-		ln.Close()
-	}
-
-	switch {
-	case err1 != nil:
-		stop()
-		return nil, nil, nil, err1
-	case err2 != nil:
-		stop()
-		return nil, nil, nil, err2
-	default:
-		return c1, c2, stop, nil
-	}
-}
-
-func Uint64() (uint64, error) {
+func RandomUint64() (uint64, error) {
 	var b [8]byte
 
 	_, err := rand.Read(b[:])

proxy/tls.go

@@ -45,6 +45,7 @@ func newTLSListenerConfig(conf *config.Config) (*tls.Config, error) {
 			return nil, errors.New("Failed to parse listener root certificate")
 		}
 		cfg.ClientCAs = clientCAs
+		cfg.ClientAuth = tls.RequireAndVerifyClientCert
 	}
 	return cfg, nil
 }