Add SASL AuthIdentity to SASL frames (authzid)

broker.go

@@ -978,10 +978,10 @@ func (b *Broker) sendAndReceiveSASLPlainAuth() error {
 // sendAndReceiveV0SASLPlainAuth flows the v0 sasl auth NOT wrapped in the kafka protocol
 func (b *Broker) sendAndReceiveV0SASLPlainAuth() error {
 
-	length := 1 + len(b.conf.Net.SASL.User) + 1 + len(b.conf.Net.SASL.Password)
+	length := len(b.conf.Net.SASL.AuthIdentity) + 1 + len(b.conf.Net.SASL.User) + 1 + len(b.conf.Net.SASL.Password)
 	authBytes := make([]byte, length+4) //4 byte length header + auth data
 	binary.BigEndian.PutUint32(authBytes, uint32(length))
-	copy(authBytes[4:], []byte("\x00"+b.conf.Net.SASL.User+"\x00"+b.conf.Net.SASL.Password))
+	copy(authBytes[4:], []byte(b.conf.Net.SASL.AuthIdentity+"\x00"+b.conf.Net.SASL.User+"\x00"+b.conf.Net.SASL.Password))
 
 	requestTime := time.Now()
 	bytesWritten, err := b.write(authBytes)
@@ -1216,7 +1216,7 @@ func mapToString(extensions map[string]string, keyValSep string, elemSep string)
 }
 
 func (b *Broker) sendSASLPlainAuthClientResponse(correlationID int32) (int, error) {
-	authBytes := []byte("\x00" + b.conf.Net.SASL.User + "\x00" + b.conf.Net.SASL.Password)
+	authBytes := []byte(b.conf.Net.SASL.AuthIdentity + "\x00" + b.conf.Net.SASL.User + "\x00" + b.conf.Net.SASL.Password)
 	rb := &SaslAuthenticateRequest{authBytes}
 	req := &request{correlationID: correlationID, clientID: b.conf.ClientID, body: rb}
 	buf, err := encode(req, b.conf.MetricRegistry)

broker_test.go

@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"net"
@@ -133,6 +134,7 @@ func TestSASLOAuthBearer(t *testing.T) {
 
 	testTable := []struct {
 		name                      string
+		authidentity              string
 		mockSASLHandshakeResponse MockResponse // Mock SaslHandshakeRequest response from broker
 		mockSASLAuthResponse      MockResponse // Mock SaslAuthenticateRequest response from broker
 		expectClientErr           bool         // Expect an internal client-side error
@@ -396,6 +398,7 @@ func TestSASLPlainAuth(t *testing.T) {
 
 	testTable := []struct {
 		name             string
+		authidentity     string
 		mockAuthErr      KError // Mock and expect error returned from SaslAuthenticateRequest
 		mockHandshakeErr KError // Mock and expect error returned from SaslHandshakeRequest
 		expectClientErr  bool   // Expect an internal client-side error
@@ -405,6 +408,12 @@ func TestSASLPlainAuth(t *testing.T) {
 			mockAuthErr:      ErrNoError,
 			mockHandshakeErr: ErrNoError,
 		},
+		{
+			name:             "SASL Plain OK server response with authidentity",
+			authidentity:     "authid",
+			mockAuthErr:      ErrNoError,
+			mockHandshakeErr: ErrNoError,
+		},
 		{
 			name:             "SASL Plain authentication failure response",
 			mockAuthErr:      ErrSASLAuthenticationFailed,
@@ -453,6 +462,7 @@ func TestSASLPlainAuth(t *testing.T) {
 
 		conf := NewConfig()
 		conf.Net.SASL.Mechanism = SASLTypePlaintext
+		conf.Net.SASL.AuthIdentity = test.authidentity
 		conf.Net.SASL.User = "token"
 		conf.Net.SASL.Password = "password"
 		conf.Net.SASL.Version = SASLHandshakeV1
@@ -474,6 +484,23 @@ func TestSASLPlainAuth(t *testing.T) {
 		broker.conn = conn
 
 		err = broker.authenticateViaSASL()
+		if err == nil {
+			for _, rr := range mockBroker.History() {
+				switch r := rr.Request.(type) {
+				case *SaslAuthenticateRequest:
+					x := bytes.SplitN(r.SaslAuthBytes, []byte("\x00"), 3)
+					if string(x[0]) != conf.Net.SASL.AuthIdentity {
+						t.Errorf("[%d]:[%s] expected %s auth identity, got %s\n", i, test.name, conf.Net.SASL.AuthIdentity, x[0])
+					}
+					if string(x[1]) != conf.Net.SASL.User {
+						t.Errorf("[%d]:[%s] expected %s user, got %s\n", i, test.name, conf.Net.SASL.User, x[1])
+					}
+					if string(x[2]) != conf.Net.SASL.Password {
+						t.Errorf("[%d]:[%s] expected %s password, got %s\n", i, test.name, conf.Net.SASL.Password, x[2])
+					}
+				}
+			}
+		}
 
 		if test.mockAuthErr != ErrNoError {
 			if test.mockAuthErr != err {

config.go

@@ -65,8 +65,15 @@ type Config struct {
 			// (defaults to true). You should only set this to false if you're using
 			// a non-Kafka SASL proxy.
 			Handshake bool
-			//username and password for SASL/PLAIN  or SASL/SCRAM authentication
-			User     string
+			// AuthIdentity is an (optional) authorization identity (authzid) to
+			// use for SASL/PLAIN authentication (if different from User) when
+			// an authenticated user is permitted to act as the presented
+			// alternative user. See RFC4616 for details.
+			AuthIdentity string
+			// User is the authentication identity (authcid) to present for
+			// SASL/PLAIN or SASL/SCRAM authentication
+			User string
+			// Password for SASL/PLAIN authentication
 			Password string
 			// authz id used for SASL/SCRAM authentication
 			SCRAMAuthzID string