feat: add SASLv1 support for Kerberos (#3279)

For some reason we were still pinning the old v0 SASL when using GSSAPI
and this doesn't work if an ApiVersionsRequest is sent before the auth
flow.
Add support for sending the krb5 bytes in the v1 SASL SaslAuthenticate
protocol wrapping.

Note: this is just a draft for now as although it is tested and working
against a GSSAPI enabled cluster, the underlying code is very much a
copy-and-paste of the existing v0 auth flow, just wrappering the bytes
in the SaslAuthenticate rather than sending them to the broker directly.
This needs more work to tidyup the implementation

---------

Signed-off-by: Dominic Evans <[email protected]>

Author: dnwe

broker.go

@@ -242,7 +242,7 @@ func (b *Broker) Open(conf *Config) error {
 			conf.Net.SASL.Version = SASLHandshakeV1
 		}
 
-		useSaslV0 := conf.Net.SASL.Version == SASLHandshakeV0 || conf.Net.SASL.Mechanism == SASLTypeGSSAPI
+		useSaslV0 := conf.Net.SASL.Version == SASLHandshakeV0
 		if conf.Net.SASL.Enable && useSaslV0 {
 			b.connErr = b.authenticateViaSASLv0()
 
@@ -1379,6 +1379,12 @@ func (b *Broker) authenticateViaSASLv1() error {
 	}
 
 	switch b.conf.Net.SASL.Mechanism {
+	case SASLTypeGSSAPI:
+		b.kerberosAuthenticator.Config = &b.conf.Net.SASL.GSSAPI
+		if b.kerberosAuthenticator.NewKerberosClientFunc == nil {
+			b.kerberosAuthenticator.NewKerberosClientFunc = NewKerberosClient
+		}
+		return b.kerberosAuthenticator.AuthorizeV2(b, authSendReceiver)
 	case SASLTypeOAuth:
 		provider := b.conf.Net.SASL.TokenProvider
 		return b.sendAndReceiveSASLOAuth(authSendReceiver, provider)

broker_test.go

@@ -666,7 +666,7 @@ func TestGSSAPIKerberosAuth_Authorize(t *testing.T) {
 		},
 		{
 			name:  "Kerberos client creation fails",
-			error: errors.New("configuration file could not be opened: krb5.conf open krb5.conf: no such file or directory"),
+			error: errors.New("configuration file could not be opened: testdata/krb5.conf open testdata/krb5.conf: no such file or directory"),
 		},
 		{
 			name:               "Bad server response, unmarshall key error",
@@ -701,10 +701,11 @@ func TestGSSAPIKerberosAuth_Authorize(t *testing.T) {
 			broker.requestsInFlight = metrics.NilCounter{}
 
 			conf := NewTestConfig()
+			conf.Net.SASL.Version = SASLHandshakeV0
 			conf.Net.SASL.Mechanism = SASLTypeGSSAPI
 			conf.Net.SASL.Enable = true
 			conf.Net.SASL.GSSAPI.ServiceName = "kafka"
-			conf.Net.SASL.GSSAPI.KerberosConfigPath = "krb5.conf"
+			conf.Net.SASL.GSSAPI.KerberosConfigPath = "testdata/krb5.conf"
 			conf.Net.SASL.GSSAPI.Realm = "EXAMPLE.COM"
 			conf.Net.SASL.GSSAPI.Username = "kafka"
 			conf.Net.SASL.GSSAPI.Password = "kafka"

config.go

@@ -655,7 +655,9 @@ func (c *Config) Validate() error {
 		if c.Net.SASL.Mechanism == "" {
 			c.Net.SASL.Mechanism = SASLTypePlaintext
 		}
-
+		if c.Net.SASL.Version == SASLHandshakeV0 && c.ApiVersionsRequest {
+			return ConfigurationError("ApiVersionsRequest must be disabled when SASL v0 is enabled")
+		}
 		switch c.Net.SASL.Mechanism {
 		case SASLTypePlaintext:
 			if c.Net.SASL.User == "" {

gssapi_kerberos.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"net"
 	"strings"
 	"time"
 
@@ -108,16 +109,14 @@ func (krbAuth *GSSAPIKerberosAuth) newAuthenticatorChecksum() []byte {
 	return a
 }
 
-/*
-*
-* Construct Kerberos AP_REQ package, conforming to RFC-4120
-* https://tools.ietf.org/html/rfc4120#page-84
-*
- */
+// Construct Kerberos AP_REQ package, conforming to RFC-4120
+// https://tools.ietf.org/html/rfc4120#page-84
 func (krbAuth *GSSAPIKerberosAuth) createKrb5Token(
-	domain string, cname types.PrincipalName,
+	domain string,
+	cname types.PrincipalName,
 	ticket messages.Ticket,
-	sessionKey types.EncryptionKey) ([]byte, error) {
+	sessionKey types.EncryptionKey,
+) ([]byte, error) {
 	auth, err := types.NewAuthenticator(domain, cname)
 	if err != nil {
 		return nil, err
@@ -144,16 +143,12 @@ func (krbAuth *GSSAPIKerberosAuth) createKrb5Token(
 	return aprBytes, nil
 }
 
-/*
-*
-*	Append the GSS-API header to the payload, conforming to RFC-2743
-*	Section 3.1, Mechanism-Independent Token Format
-*
-*	https://tools.ietf.org/html/rfc2743#page-81
-*
-*	GSSAPIHeader + <specific mechanism payload>
-*
- */
+// Append the GSS-API header to the payload, conforming to RFC-2743
+// Section 3.1, Mechanism-Independent Token Format
+//
+// https://tools.ietf.org/html/rfc2743#page-81
+//
+// GSSAPIHeader + <specific mechanism payload>
 func (krbAuth *GSSAPIKerberosAuth) appendGSSAPIHeader(payload []byte) ([]byte, error) {
 	oidBytes, err := asn1.Marshal(gssapi.OIDKRB5.OID())
 	if err != nil {
@@ -166,12 +161,15 @@ func (krbAuth *GSSAPIKerberosAuth) appendGSSAPIHeader(payload []byte) ([]byte, e
 	return GSSPackage, nil
 }
 
-func (krbAuth *GSSAPIKerberosAuth) initSecContext(bytes []byte, kerberosClient KerberosClient) ([]byte, error) {
+func (krbAuth *GSSAPIKerberosAuth) initSecContext(
+	client KerberosClient,
+	bytes []byte,
+) ([]byte, error) {
 	switch krbAuth.step {
 	case GSS_API_INITIAL:
 		aprBytes, err := krbAuth.createKrb5Token(
-			kerberosClient.Domain(),
-			kerberosClient.CName(),
+			client.Domain(),
+			client.CName(),
 			krbAuth.ticket,
 			krbAuth.encKey)
 		if err != nil {
@@ -200,63 +198,123 @@ func (krbAuth *GSSAPIKerberosAuth) initSecContext(bytes []byte, kerberosClient K
 	return nil, nil
 }
 
-/* This does the handshake for authorization */
-func (krbAuth *GSSAPIKerberosAuth) Authorize(broker *Broker) error {
-	kerberosClient, err := krbAuth.NewKerberosClientFunc(krbAuth.Config)
-	if err != nil {
-		Logger.Printf("Kerberos client error: %s", err)
-		return err
-	}
-
-	err = kerberosClient.Login()
-	if err != nil {
-		Logger.Printf("Kerberos client error: %s", err)
-		return err
-	}
-	// Construct SPN using serviceName and host
-	// default SPN format: <SERVICE>/<FQDN>
-
-	host := strings.SplitN(broker.addr, ":", 2)[0] // Strip port part
+func (krbAuth *GSSAPIKerberosAuth) spn(broker *Broker) string {
+	host, _, _ := net.SplitHostPort(broker.addr)
 	var spn string
 	if krbAuth.Config.BuildSpn != nil {
 		spn = krbAuth.Config.BuildSpn(broker.conf.Net.SASL.GSSAPI.ServiceName, host)
 	} else {
 		spn = fmt.Sprintf("%s/%s", broker.conf.Net.SASL.GSSAPI.ServiceName, host)
 	}
+	return spn
+}
 
-	ticket, encKey, err := kerberosClient.GetServiceTicket(spn)
+// Login will use the given KerberosClient to login and get a ticket for the given spn.
+func (krbAuth *GSSAPIKerberosAuth) Login(
+	client KerberosClient,
+	spn string,
+) (*messages.Ticket, error) {
+	if err := client.Login(); err != nil {
+		Logger.Printf("Kerberos client login error: %s", err)
+		return nil, err
+	}
+
+	ticket, encKey, err := client.GetServiceTicket(spn)
 	if err != nil {
-		Logger.Printf("Error getting Kerberos service ticket : %s", err)
-		return err
+		Logger.Printf("Kerberos service ticket error for %s: %s", spn, err)
+		return nil, err
 	}
 	krbAuth.ticket = ticket
 	krbAuth.encKey = encKey
 	krbAuth.step = GSS_API_INITIAL
-	var receivedBytes []byte = nil
-	defer kerberosClient.Destroy()
+
+	return &ticket, nil
+}
+
+// Authorize performs the kerberos auth handshake for authorization
+func (krbAuth *GSSAPIKerberosAuth) Authorize(broker *Broker) error {
+	client, err := krbAuth.NewKerberosClientFunc(krbAuth.Config)
+	if err != nil {
+		Logger.Printf("Kerberos client initialization error: %s", err)
+		return err
+	}
+	defer client.Destroy()
+
+	ticket, err := krbAuth.Login(client, krbAuth.spn(broker))
+	if err != nil {
+		return err
+	}
+
+	principal := strings.Join(ticket.SName.NameString, "/") + "@" + ticket.Realm
+	var receivedBytes []byte
+
 	for {
-		packBytes, err := krbAuth.initSecContext(receivedBytes, kerberosClient)
+		packBytes, err := krbAuth.initSecContext(client, receivedBytes)
 		if err != nil {
-			Logger.Printf("Error while performing GSSAPI Kerberos Authentication: %s\n", err)
+			Logger.Printf("Kerberos init error as %s: %s", principal, err)
 			return err
 		}
+
 		requestTime := time.Now()
 		bytesWritten, err := krbAuth.writePackage(broker, packBytes)
 		if err != nil {
-			Logger.Printf("Error while performing GSSAPI Kerberos Authentication: %s\n", err)
+			Logger.Printf("Kerberos write error as %s: %s", principal, err)
 			return err
 		}
 		broker.updateOutgoingCommunicationMetrics(bytesWritten)
-		if krbAuth.step == GSS_API_VERIFY {
-			bytesRead := 0
+
+		switch krbAuth.step {
+		case GSS_API_VERIFY:
+			var bytesRead int
 			receivedBytes, bytesRead, err = krbAuth.readPackage(broker)
 			requestLatency := time.Since(requestTime)
 			broker.updateIncomingCommunicationMetrics(bytesRead, requestLatency)
 			if err != nil {
-				Logger.Printf("Error while performing GSSAPI Kerberos Authentication: %s\n", err)
+				Logger.Printf("Kerberos read error as %s: %s", principal, err)
 				return err
 			}
-		} else if krbAuth.step == GSS_API_FINISH {
+		case GSS_API_FINISH:
+			return nil
+		}
+	}
+}
+
+// AuthorizeV2 performs the SASL v2 GSSAPI authentication with the Kafka broker.
+func (krbAuth *GSSAPIKerberosAuth) AuthorizeV2(
+	broker *Broker,
+	authSendReceiver func(authBytes []byte) (*SaslAuthenticateResponse, error),
+) error {
+	client, err := krbAuth.NewKerberosClientFunc(krbAuth.Config)
+	if err != nil {
+		Logger.Printf("Kerberos client initialization error: %s", err)
+		return err
+	}
+	defer client.Destroy()
+
+	ticket, err := krbAuth.Login(client, krbAuth.spn(broker))
+	if err != nil {
+		return err
+	}
+
+	principal := strings.Join(ticket.SName.NameString, "/") + "@" + ticket.Realm
+	var receivedBytes []byte
+
+	for {
+		token, err := krbAuth.initSecContext(client, receivedBytes)
+		if err != nil {
+			Logger.Printf("SASL Kerberos init error as %s: %s", principal, err)
+			return err
+		}
+
+		authResponse, err := authSendReceiver(token)
+		if err != nil {
+			Logger.Printf("SASL Kerberos authenticate error as %s: %s", principal, err)
+			return err
+		}
+
+		receivedBytes = authResponse.SaslAuthBytes
+
+		if krbAuth.step == GSS_API_FINISH {
 			return nil
 		}
 	}

kerberos_client_test.go

@@ -60,7 +60,7 @@ const (
 )
 
 func TestFaildToCreateKerberosConfig(t *testing.T) {
-	expectedErr := errors.New("configuration file could not be opened: krb5.conf open krb5.conf: no such file or directory")
+	expectedErr := errors.New("configuration file could not be opened: testdata/krb5.conf open testdata/krb5.conf: no such file or directory")
 	clientConfig := NewTestConfig()
 	clientConfig.Net.SASL.Mechanism = SASLTypeGSSAPI
 	clientConfig.Net.SASL.Enable = true
@@ -69,7 +69,7 @@ func TestFaildToCreateKerberosConfig(t *testing.T) {
 	clientConfig.Net.SASL.GSSAPI.Username = "client"
 	clientConfig.Net.SASL.GSSAPI.AuthType = KRB5_USER_AUTH
 	clientConfig.Net.SASL.GSSAPI.Password = "qwerty"
-	clientConfig.Net.SASL.GSSAPI.KerberosConfigPath = "krb5.conf"
+	clientConfig.Net.SASL.GSSAPI.KerberosConfigPath = "testdata/krb5.conf"
 	_, err := NewKerberosClient(&clientConfig.Net.SASL.GSSAPI)
 	// Expect to create client with password
 	if err.Error() != expectedErr.Error() {