Implement client certificate subject validation

cmd/kafka-proxy/server.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"fmt"
+
 	"github.com/grepplabs/kafka-proxy/config"
 	"github.com/grepplabs/kafka-proxy/proxy"
 	"github.com/oklog/run"
@@ -20,13 +21,14 @@ import (
 	"time"
 
 	"errors"
+	"strings"
+
 	"github.com/grepplabs/kafka-proxy/pkg/apis"
 	localauth "github.com/grepplabs/kafka-proxy/plugin/local-auth/shared"
 	tokeninfo "github.com/grepplabs/kafka-proxy/plugin/token-info/shared"
 	tokenprovider "github.com/grepplabs/kafka-proxy/plugin/token-provider/shared"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-plugin"
-	"strings"
 
 	"github.com/grepplabs/kafka-proxy/pkg/registry"
 	// built-in plugins
@@ -105,6 +107,14 @@ func initFlags() {
 	Server.Flags().StringSliceVar(&c.Proxy.TLS.ListenerCipherSuites, "proxy-listener-cipher-suites", []string{}, "List of supported cipher suites")
 	Server.Flags().StringSliceVar(&c.Proxy.TLS.ListenerCurvePreferences, "proxy-listener-curve-preferences", []string{}, "List of curve preferences")
 
+	Server.Flags().BoolVar(&c.Proxy.TLS.ClientCert.ValidateSubject, "proxy-listener-tls-client-cert-validate-subject", false, "Whether to validate client certificate subject")
+	Server.Flags().StringVar(&c.Proxy.TLS.ClientCert.Subject.CommonName, "proxy-listener-tls-client-cert-subject-common-name", "", "Required client certificate subject common name")
+	Server.Flags().StringSliceVar(&c.Proxy.TLS.ClientCert.Subject.Country, "proxy-listener-tls-required-client-subject-country", []string{}, "Required client certificate subject country")
+	Server.Flags().StringSliceVar(&c.Proxy.TLS.ClientCert.Subject.Province, "proxy-listener-tls-required-client-subject-province", []string{}, "Required client certificate subject province")
+	Server.Flags().StringSliceVar(&c.Proxy.TLS.ClientCert.Subject.Locality, "proxy-listener-tls-required-client-subject-locality", []string{}, "Required client certificate subject locality")
+	Server.Flags().StringSliceVar(&c.Proxy.TLS.ClientCert.Subject.Organization, "proxy-listener-tls-required-client-subject-organization", []string{}, "Required client certificate subject organization")
+	Server.Flags().StringSliceVar(&c.Proxy.TLS.ClientCert.Subject.OrganizationalUnit, "proxy-listener-tls-required-client-subject-organizational-unit", []string{}, "Required client certificate subject organizational unit")
+
 	// 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")
 	Server.Flags().StringVar(&c.Auth.Local.Command, "auth-local-command", "", "Path to authentication plugin binary")

config/config.go

@@ -2,12 +2,13 @@ package config
 
 import (
 	"fmt"
-	"github.com/grepplabs/kafka-proxy/pkg/libs/util"
-	"github.com/pkg/errors"
 	"net"
 	"net/url"
 	"strings"
 	"time"
+
+	"github.com/grepplabs/kafka-proxy/pkg/libs/util"
+	"github.com/pkg/errors"
 )
 
 const defaultClientID = "kafka-proxy"
@@ -70,6 +71,17 @@ type Config struct {
 			CAChainCertFile          string
 			ListenerCipherSuites     []string
 			ListenerCurvePreferences []string
+			ClientCert               struct {
+				ValidateSubject bool
+				Subject         struct {
+					CommonName         string
+					Country            []string
+					Province           []string
+					Locality           []string
+					Organization       []string
+					OrganizationalUnit []string
+				}
+			}
 		}
 	}
 	Auth struct {

proxy/proxy.go

@@ -3,11 +3,12 @@ package proxy
 import (
 	"crypto/tls"
 	"fmt"
+	"net"
+	"sync"
+
 	"github.com/grepplabs/kafka-proxy/config"
 	"github.com/grepplabs/kafka-proxy/pkg/libs/util"
 	"github.com/sirupsen/logrus"
-	"net"
-	"sync"
 )
 
 type ListenFunc func(cfg config.ListenerConfig) (l net.Listener, err error)

proxy/tls.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/pem"
+	"fmt"
 	"io/ioutil"
 	"net"
 	"strings"
@@ -120,6 +121,35 @@ func newTLSListenerConfig(conf *config.Config) (*tls.Config, error) {
 		cfg.ClientCAs = clientCAs
 		cfg.ClientAuth = tls.RequireAndVerifyClientCert
 	}
+
+	cfg.VerifyPeerCertificate = func(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
+		if conf.Proxy.TLS.ClientCert.ValidateSubject {
+			expected := fmt.Sprintf("s:/CN=%s/C=%v/S=%v/L=%v/O=%v/OU=%v",
+				conf.Proxy.TLS.ClientCert.Subject.CommonName,
+				conf.Proxy.TLS.ClientCert.Subject.Country,
+				conf.Proxy.TLS.ClientCert.Subject.Province,
+				conf.Proxy.TLS.ClientCert.Subject.Locality,
+				conf.Proxy.TLS.ClientCert.Subject.Organization,
+				conf.Proxy.TLS.ClientCert.Subject.OrganizationalUnit)
+			for _, chain := range verifiedChains {
+				for _, cert := range chain {
+					current := fmt.Sprintf("s:/CN=%s/C=%v/S=%v/L=%v/O=%v/OU=%v",
+						cert.Subject.CommonName,
+						cert.Subject.Country,
+						cert.Subject.Province,
+						cert.Subject.Locality,
+						cert.Subject.Organization,
+						cert.Subject.OrganizationalUnit)
+					if current == expected {
+						return nil
+					}
+				}
+			}
+			return fmt.Errorf("tls: no client certificate presented required subject '%s'", expected)
+		}
+		return nil
+	}
+
 	return cfg, nil
 }
 

proxy/tls_test.go

@@ -3,16 +3,18 @@ package proxy
 import (
 	"bytes"
 	"crypto/x509"
-	"github.com/armon/go-socks5"
-	"github.com/grepplabs/kafka-proxy/config"
-	"github.com/pkg/errors"
-	"github.com/stretchr/testify/assert"
+	"crypto/x509/pkix"
 	"io"
 	"net"
 	"os"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/armon/go-socks5"
+	"github.com/grepplabs/kafka-proxy/config"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestDefaultCipherSuites(t *testing.T) {
@@ -51,6 +53,69 @@ func TestEnabledCipherSuites(t *testing.T) {
 	a.Equal(1, len(serverConfig.CurvePreferences))
 }
 
+func TestValidEnabledClientCertSubjectValidate(t *testing.T) {
+	a := assert.New(t)
+	testSubject := pkix.Name{
+		CommonName:         "integration-test",
+		Country:            []string{"DE"},
+		Locality:           []string{"test-file"},
+		Organization:       []string{"integration-test"},
+		OrganizationalUnit: []string{"invalid-OrganizationalUnit"},
+	}
+	bundle := NewCertsBundleWithSubject(testSubject)
+	defer bundle.Close()
+	c := new(config.Config)
+	c.Proxy.TLS.ClientCert.ValidateSubject = true
+	c.Proxy.TLS.ClientCert.Subject.CommonName = testSubject.CommonName
+	c.Proxy.TLS.ClientCert.Subject.Country = testSubject.Country
+	c.Proxy.TLS.ClientCert.Subject.Locality = testSubject.Locality
+	c.Proxy.TLS.ClientCert.Subject.Organization = testSubject.Organization
+	c.Proxy.TLS.ClientCert.Subject.OrganizationalUnit = testSubject.OrganizationalUnit
+	c.Proxy.TLS.ListenerCertFile = bundle.ServerCert.Name()
+	c.Proxy.TLS.ListenerKeyFile = bundle.ServerKey.Name()
+	c.Proxy.TLS.CAChainCertFile = bundle.CACert.Name()
+
+	c.Kafka.TLS.CAChainCertFile = bundle.CACert.Name()
+	c.Kafka.TLS.ClientCertFile = bundle.ClientCert.Name()
+	c.Kafka.TLS.ClientKeyFile = bundle.ClientKey.Name()
+
+	_, _, _, err := makeTLSPipe(c, nil)
+
+	a.Nil(err)
+}
+
+func TestInvalidEnabledClientCertSubjectValidate(t *testing.T) {
+	a := assert.New(t)
+	testSubject := pkix.Name{
+		CommonName:         "integration-test",
+		Country:            []string{"DE"},
+		Locality:           []string{"test-file"},
+		Organization:       []string{"integration-test"},
+		OrganizationalUnit: []string{"invalid-OrganizationalUnit"},
+	}
+	bundle := NewCertsBundleWithSubject(testSubject)
+	defer bundle.Close()
+	c := new(config.Config)
+	c.Proxy.TLS.ClientCert.ValidateSubject = true
+	c.Proxy.TLS.ClientCert.Subject.CommonName = testSubject.CommonName
+	c.Proxy.TLS.ClientCert.Subject.Country = testSubject.Country
+	c.Proxy.TLS.ClientCert.Subject.Locality = testSubject.Locality
+	c.Proxy.TLS.ClientCert.Subject.Organization = testSubject.Organization
+	c.Proxy.TLS.ClientCert.Subject.OrganizationalUnit = []string{"expected-OrganizationalUnit"}
+	c.Proxy.TLS.ListenerCertFile = bundle.ServerCert.Name()
+	c.Proxy.TLS.ListenerKeyFile = bundle.ServerKey.Name()
+	c.Proxy.TLS.CAChainCertFile = bundle.CACert.Name()
+
+	c.Kafka.TLS.CAChainCertFile = bundle.CACert.Name()
+	c.Kafka.TLS.ClientCertFile = bundle.ClientCert.Name()
+	c.Kafka.TLS.ClientKeyFile = bundle.ClientKey.Name()
+
+	_, _, _, err := makeTLSPipe(c, nil)
+
+	a.NotNil(err)
+	a.Contains(err.Error(), "tls: no client certificate presented required subject 's:/CN=integration-test/C=[DE]/S=[]/L=[test-file]/O=[integration-test]/OU=[expected-OrganizationalUnit]'")
+}
+
 func TestTLSUnknownAuthorityNoCAChainCert(t *testing.T) {
 	a := assert.New(t)
 

proxy/util_test.go

@@ -7,18 +7,19 @@ import (
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/pem"
-	"github.com/armon/go-socks5"
-	"github.com/elazarl/goproxy"
-	"github.com/elazarl/goproxy/ext/auth"
-	"github.com/grepplabs/kafka-proxy/config"
-	"github.com/pkg/errors"
-	"golang.org/x/net/proxy"
 	"io/ioutil"
 	"math/big"
 	"net"
 	"net/http"
 	"os"
 	"time"
+
+	"github.com/armon/go-socks5"
+	"github.com/elazarl/goproxy"
+	"github.com/elazarl/goproxy/ext/auth"
+	"github.com/grepplabs/kafka-proxy/config"
+	"github.com/pkg/errors"
+	"golang.org/x/net/proxy"
 )
 
 type testAcceptResult struct {
@@ -380,18 +381,23 @@ func makeHttpProxyPipe() (net.Conn, net.Conn, func(), error) {
 }
 
 func generateCert(catls *tls.Certificate, certFile *os.File, keyFile *os.File) error {
+	return generateCertWithSubject(catls, certFile, keyFile, pkix.Name{
+		Organization:       []string{"ORGANIZATION_NAME"},
+		OrganizationalUnit: []string{"ORGANIZATIONAL_UNIT"},
+		Country:            []string{"COUNTRY_CODE"},
+		Province:           []string{"PROVINCE"},
+		Locality:           []string{"CITY"},
+		StreetAddress:      []string{"ADDRESS"},
+		PostalCode:         []string{"POSTAL_CODE"},
+		CommonName:         "localhost",
+	})
+}
+
+func generateCertWithSubject(catls *tls.Certificate, certFile *os.File, keyFile *os.File, subject pkix.Name) error {
 	// Prepare certificate
 	cert := &x509.Certificate{
 		SerialNumber: big.NewInt(1),
-		Subject: pkix.Name{
-			Organization:  []string{"ORGANIZATION_NAME"},
-			Country:       []string{"COUNTRY_CODE"},
-			Province:      []string{"PROVINCE"},
-			Locality:      []string{"CITY"},
-			StreetAddress: []string{"ADDRESS"},
-			PostalCode:    []string{"POSTAL_CODE"},
-			CommonName:    "localhost",
-		},
+		Subject:      subject,
 		NotBefore:    time.Now(),
 		NotAfter:     time.Now().AddDate(10, 0, 0),
 		SubjectKeyId: []byte{1, 2, 3, 4, 6},
@@ -539,6 +545,52 @@ func NewCertsBundle() *CertsBundle {
 	return bundle
 }
 
+func NewCertsBundleWithSubject(subject pkix.Name) *CertsBundle {
+	bundle := &CertsBundle{}
+	dirName, err := ioutil.TempDir("", "tls-test")
+	if err != nil {
+		panic(err)
+	}
+	bundle.CACert, err = ioutil.TempFile(dirName, "ca-cert-")
+	if err != nil {
+		panic(err)
+	}
+	bundle.CAKey, err = ioutil.TempFile(dirName, "ca-key-")
+	if err != nil {
+		panic(err)
+	}
+	bundle.ServerCert, err = ioutil.TempFile(dirName, "server-cert-")
+	if err != nil {
+		panic(err)
+	}
+	bundle.ServerKey, err = ioutil.TempFile(dirName, "server-key-")
+	if err != nil {
+		panic(err)
+	}
+	bundle.ClientCert, err = ioutil.TempFile(dirName, "client-cert-")
+	if err != nil {
+		panic(err)
+	}
+	bundle.ClientKey, err = ioutil.TempFile("", "client-key-")
+	if err != nil {
+		panic(err)
+	}
+	// generate certs
+	catls, err := generateCA(bundle.CACert, bundle.CAKey)
+	if err != nil {
+		panic(err)
+	}
+	err = generateCert(catls, bundle.ServerCert, bundle.ServerKey)
+	if err != nil {
+		panic(err)
+	}
+	err = generateCertWithSubject(catls, bundle.ClientCert, bundle.ClientKey, subject)
+	if err != nil {
+		panic(err)
+	}
+	return bundle
+}
+
 func (bundle *CertsBundle) Close() {
 	_ = os.Remove(bundle.CACert.Name())
 	_ = os.Remove(bundle.CAKey.Name())