Add kerberos support

Signed-off-by: Ruben Vargas <ruben.vp8510@gmail.com>

broker.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"encoding/binary"
 	"fmt"
+	"github.com/rcrowley/go-metrics"
 	"io"
 	"net"
 	"sort"
@@ -12,7 +13,6 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
-
 	metrics "github.com/rcrowley/go-metrics"
 )
 
@@ -61,6 +61,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 +845,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.conn)
+}
+
 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,158 @@
+package sarama
+
+import (
+	"encoding/binary"
+	"encoding/hex"
+	"github.com/gokrb5/iana/keyusage"
+	"github.com/jcmturner/gofork/encoding/asn1"
+	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"
+	"net"
+)
+
+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(conn net.Conn, 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))
+	conn.Write(finalPackage)
+}
+
+func (krbAuth *GSSAPIKerberosAuth) readPackage(conn net.Conn) []byte {
+	lengthInBytes := make([]byte, 4)
+	io.ReadFull(conn, lengthInBytes)
+	payloadLength := binary.BigEndian.Uint32(lengthInBytes)
+	payloadBytes := make([]byte, payloadLength) // buffer for read..
+	io.ReadFull(conn, payloadBytes)             // read bytes
+	return payloadBytes
+}
+
+
+func (krbAuth *GSSAPIKerberosAuth) gssapi_encodeLength(len int32) []byte {
+	var bytes []byte
+	if len < 128 {
+		return append(bytes, byte(len))
+	} else if len < (1 << 8) {
+		return append(bytes, 0x081, byte(len))
+	} else if len < (1 << 16) {
+		return append(bytes, 0x082, byte(len>>8), byte(len))
+	} else if len < (1 << 24) {
+		return append(bytes, 0x083, byte(len>>16), byte(len>>8), byte(len))
+	}
+	return append(bytes, 0x084, byte(len>>24), byte(len>>16), byte(len>>8), byte(len))
+}
+
+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 := krbAuth.gssapi_encodeLength(int32(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(conn net.Conn) 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(conn, GSSPackage)
+	wrapBytes := krbAuth.readPackage(conn)
+	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(conn, wrapResponseBytes)
+	return nil
+}