feat: experimental SASLv1 support for Kerberos

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.

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()
 
@@ -1370,6 +1370,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

@@ -117,7 +117,8 @@ func (krbAuth *GSSAPIKerberosAuth) newAuthenticatorChecksum() []byte {
 func (krbAuth *GSSAPIKerberosAuth) createKrb5Token(
 	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
@@ -200,63 +201,122 @@ 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 := strings.SplitN(broker.addr, ":", 2)[0]
 	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
+}
+
+// Login will use the given KerberosClient to login and get a ticket for the given spn.
+func (krbAuth *GSSAPIKerberosAuth) Login(kerberosClient KerberosClient, spn string) (*messages.Ticket, error) {
+	if err := kerberosClient.Login(); err != nil {
+		Logger.Printf("Kerberos client login error: %s", err)
+		return nil, err
+	}
 
 	ticket, encKey, err := kerberosClient.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
+
+	return &ticket, nil
+}
+
+// Authorize performs the kerberos auth handshake for authorization
+func (krbAuth *GSSAPIKerberosAuth) Authorize(broker *Broker) error {
+	kerberosClient, err := krbAuth.NewKerberosClientFunc(krbAuth.Config)
+	if err != nil {
+		Logger.Printf("Kerberos client initialization error: %s", err)
+		return err
+	}
 	defer kerberosClient.Destroy()
+
+	ticket, err := krbAuth.Login(kerberosClient, 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)
 		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 {
+	kerberosClient, err := krbAuth.NewKerberosClientFunc(krbAuth.Config)
+	if err != nil {
+		Logger.Printf("Kerberos client initialization error: %s", err)
+		return err
+	}
+	defer kerberosClient.Destroy()
+
+	ticket, err := krbAuth.Login(kerberosClient, krbAuth.spn(broker))
+	if err != nil {
+		return err
+	}
+
+	principal := strings.Join(ticket.SName.NameString, "/") + "@" + ticket.Realm
+	var receivedBytes []byte
+
+	for {
+		token, err := krbAuth.initSecContext(receivedBytes, kerberosClient)
+		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
+		}
+
+		if authResponse.Err != ErrNoError {
+			Logger.Printf("SASL Kerberos authentication failed as %s: %s", principal, authResponse.Err)
+			return authResponse.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() {