Add kerberos support

Signed-off-by: Ruben Vargas <[email protected]>

Author: rubenvp8510

broker.go

@@ -12,7 +12,6 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
-
 	metrics "github.com/rcrowley/go-metrics"
 )
 
@@ -61,6 +60,7 @@ const (
 	SASLTypeSCRAMSHA256 = "SCRAM-SHA-256"
 	// SASLTypeSCRAMSHA512 represents the SCRAM-SHA-512 mechanism.
 	SASLTypeSCRAMSHA512 = "SCRAM-SHA-512"
+	SASLTypeGSSAPI = "GSSAPI"
 	// SASLHandshakeV0 is v0 of the Kafka SASL handshake protocol. Client and
 	// server negotiate SASL auth using opaque packets.
 	SASLHandshakeV0 = int16(0)
@@ -844,11 +844,17 @@ func (b *Broker) authenticateViaSASL() error {
 		return b.sendAndReceiveSASLOAuth(b.conf.Net.SASL.TokenProvider)
 	case SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512:
 		return b.sendAndReceiveSASLSCRAMv1()
+	case SASLTypeGSSAPI:
+		return b.sendAndReceiveKerberos()
 	default:
 		return b.sendAndReceiveSASLPlainAuth()
 	}
 }
 
+func (b *Broker) sendAndReceiveKerberos() error{
+	return NewGSSAPIKerberosAuthenticator(&b.conf.Net.SASL.GSSAPI).Authorize(b)
+}
+
 func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int16) error {
 	rb := &SaslHandshakeRequest{Mechanism: string(saslType), Version: version}
 

config.go

@@ -75,6 +75,8 @@ type Config struct {
 			// AccessTokenProvider interface docs for proper implementation
 			// guidelines.
 			TokenProvider AccessTokenProvider
+
+			GSSAPI GSSAPIConfig
 		}
 
 		// KeepAlive specifies the keep-alive period for an active network connection.
@@ -520,6 +522,8 @@ func (c *Config) Validate() error {
 			if c.Net.SASL.SCRAMClientGeneratorFunc == nil {
 				return ConfigurationError("A SCRAMClientGeneratorFunc function must be provided to Net.SASL.SCRAMClientGeneratorFunc")
 			}
+		case SASLTypeGSSAPI:
+			break
 		default:
 			msg := fmt.Sprintf("The SASL mechanism configuration is invalid. Possible values are `%s`, `%s`, `%s` and `%s`",
 				SASLTypeOAuth, SASLTypePlaintext, SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512)

gssapi_kerberos.go

@@ -0,0 +1,147 @@
+package sarama
+
+import (
+	"encoding/binary"
+	"encoding/hex"
+	"github.com/gokrb5/iana/keyusage"
+	"github.com/jcmturner/gofork/encoding/asn1"
+	"gopkg.in/jcmturner/gokrb5.v7/asn1tools"
+	krbclient "gopkg.in/jcmturner/gokrb5.v7/client"
+	krbconfig "gopkg.in/jcmturner/gokrb5.v7/config"
+	"gopkg.in/jcmturner/gokrb5.v7/gssapi"
+	"gopkg.in/jcmturner/gokrb5.v7/iana/chksumtype"
+	"gopkg.in/jcmturner/gokrb5.v7/keytab"
+	"gopkg.in/jcmturner/gokrb5.v7/messages"
+	"gopkg.in/jcmturner/gokrb5.v7/types"
+	"io"
+)
+
+const (
+	TOK_ID_KRB_AP_REQ   = "0100"
+	GSS_API_GENERIC_TAG = 0x60
+)
+
+type GSSAPIConfig struct {
+	KeyTabPath         string
+	KerberosConfigPath string
+	SPN                string
+	Username           string
+	Realm              string
+}
+
+type GSSAPIKerberosAuth struct {
+	config        *GSSAPIConfig
+	client        *krbclient.Client
+	ticket        messages.Ticket
+	encryptionKey types.EncryptionKey
+}
+
+func NewGSSAPIKerberosAuthenticator(config *GSSAPIConfig) *GSSAPIKerberosAuth {
+	return &GSSAPIKerberosAuth{
+		config: config,
+	}
+}
+
+func (krbAuth *GSSAPIKerberosAuth) writePackage(broker *Broker, payload []byte) {
+	length := len(payload)
+	finalPackage := make([]byte, length+4) //4 byte length header + payload
+	copy(finalPackage[4:], payload)
+	binary.BigEndian.PutUint32(finalPackage, uint32(length))
+	bytes, err := broker.conn.Write(finalPackage)
+	broker.updateOutgoingCommunicationMetrics(bytes)
+	if err != nil {
+		Logger.Printf("Failed to send GSS-API Kerberos token: %s\n", err.Error())
+	}
+}
+
+func (krbAuth *GSSAPIKerberosAuth) readPackage(broker *Broker) []byte {
+	lengthInBytes := make([]byte, 4)
+	io.ReadFull(broker.conn, lengthInBytes)
+	payloadLength := binary.BigEndian.Uint32(lengthInBytes)
+	payloadBytes := make([]byte, payloadLength) // buffer for read..
+	io.ReadFull(broker.conn, payloadBytes)      // read bytes
+	return payloadBytes
+}
+
+func (krbAuth *GSSAPIKerberosAuth) newAuthenticatorChecksum(flags []int) []byte {
+	a := make([]byte, 24)
+	binary.LittleEndian.PutUint32(a[:4], 16)
+	for _, i := range flags {
+		if i == gssapi.ContextFlagDeleg {
+			x := make([]byte, 28-len(a))
+			a = append(a, x...)
+		}
+		f := binary.LittleEndian.Uint32(a[20:24])
+		f |= uint32(i)
+		binary.LittleEndian.PutUint32(a[20:24], f)
+	}
+	return a
+}
+
+func (krbAuth *GSSAPIKerberosAuth) createKrb5Token(client *krbclient.Client, ticket messages.Ticket, sessionKey types.EncryptionKey) []byte {
+	var GSSAPIFlags = []int{gssapi.ContextFlagInteg, gssapi.ContextFlagConf}
+	auth, _ := types.NewAuthenticator(client.Credentials.Domain(), client.Credentials.CName())
+	auth.Cksum = types.Checksum{
+		CksumType: chksumtype.GSSAPI,
+		Checksum:  krbAuth.newAuthenticatorChecksum(GSSAPIFlags),
+	}
+	APReq, _ := messages.NewAPReq(
+		ticket,
+		sessionKey,
+		auth,
+	)
+	aprBytes, _ := hex.DecodeString(TOK_ID_KRB_AP_REQ)
+	tb, _ := APReq.Marshal()
+	aprBytes = append(aprBytes, tb...)
+	return aprBytes
+}
+
+func (krbAuth *GSSAPIKerberosAuth) appendGSSAPIHeader(payload []byte) []byte {
+	oidBytes, _ := asn1.Marshal(gssapi.OID(gssapi.OIDKRB5))
+	tkoLengthBytes := asn1tools.MarshalLengthBytes(len(oidBytes) + len(payload))
+	GSSHeader := append([]byte{GSS_API_GENERIC_TAG}, tkoLengthBytes...)
+	GSSHeader = append(GSSHeader, oidBytes...)
+	GSSPackage := append(GSSHeader, payload...)
+	return GSSPackage
+}
+
+func (krbAuth *GSSAPIKerberosAuth) createKerberosClient() error {
+	kt, _ := keytab.Load(krbAuth.config.KeyTabPath)
+	cfg, _ := krbconfig.Load(krbAuth.config.KerberosConfigPath)
+	krbAuth.client = krbclient.NewClientWithKeytab(krbAuth.config.Username, krbAuth.config.Realm, kt, cfg)
+	return nil
+}
+
+func (krbAuth *GSSAPIKerberosAuth) Authorize(broker *Broker) error {
+
+	krbAuth.createKerberosClient()
+	ticket, encryptionKey, err := krbAuth.client.GetServiceTicket(krbAuth.config.SPN)
+	if err != nil {
+		print(err)
+	}
+	krbAuth.ticket = ticket
+	krbAuth.encryptionKey = encryptionKey
+	aprBytes := krbAuth.createKrb5Token(krbAuth.client, krbAuth.ticket, krbAuth.encryptionKey)
+	GSSPackage := krbAuth.appendGSSAPIHeader(aprBytes)
+	krbAuth.writePackage(broker, GSSPackage)
+	wrapBytes := krbAuth.readPackage(broker)
+	wrapTokenReq := gssapi.WrapToken{}
+	err = wrapTokenReq.Unmarshal(wrapBytes, true)
+	if err != nil {
+		Logger.Printf("Error while performing Kerberos Authentication: %s\n", err)
+		return err
+	}
+
+	// Validate response.
+	isValid, _ := wrapTokenReq.Verify(krbAuth.encryptionKey, keyusage.GSSAPI_ACCEPTOR_SEAL)
+	if !isValid {
+		Logger.Printf("Invalid key")
+
+	}
+
+	// Is valid response, reply..
+	wrapTokenResponse, _ := gssapi.NewInitiatorWrapToken(wrapTokenReq.Payload, krbAuth.encryptionKey)
+	wrapResponseBytes, _ := wrapTokenResponse.Marshal()
+	krbAuth.writePackage(broker, wrapResponseBytes)
+	return nil
+}