Merge pull request #117 from grepplabs/improve-gssapi-auth

Improve gssapi auth

.travis.yml

@@ -3,7 +3,7 @@ language: go
 dist: focal
 
 go:
-  - "1.17.x"
+  - "1.19.x"
 
 env:
   global:

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.17-alpine3.14 as builder
+FROM golang:1.19-alpine3.15 as builder
 RUN apk add alpine-sdk ca-certificates
 
 WORKDIR /go/src/github.com/grepplabs/kafka-proxy

Makefile

@@ -10,7 +10,7 @@ VERSION       ?= $(shell git describe --tags --always --dirty)
 GOPKGS         = $(shell go list ./... | grep -v /vendor/)
 BUILD_FLAGS   ?=
 LDFLAGS       ?= -X github.com/grepplabs/kafka-proxy/config.Version=$(VERSION) -w -s
-TAG           ?= "v0.3.2"
+TAG           ?= "v0.3.3"
 GOARCH        ?= amd64
 GOOS          ?= linux
 

cmd/kafka-proxy/server.go

@@ -164,15 +164,18 @@ func initFlags() {
 	Server.Flags().StringVar(&c.Kafka.SASL.Username, "sasl-username", "", "SASL user name")
 	Server.Flags().StringVar(&c.Kafka.SASL.Password, "sasl-password", "", "SASL user password")
 	Server.Flags().StringVar(&c.Kafka.SASL.JaasConfigFile, "sasl-jaas-config-file", "", "Location of JAAS config file with SASL username and password")
-	Server.Flags().StringVar(&c.Kafka.SASL.Method, "sasl-method", "PLAIN", "SASL method to use (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512")
+	Server.Flags().StringVar(&c.Kafka.SASL.Method, "sasl-method", "PLAIN", "SASL method to use (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI")
 
 	// SASL GSSAPI
-	Server.Flags().BoolVar(&c.Kafka.GSSAPI.Enable, "gssapi-enable", false, "Connect using SASL_GSSAPI")
-	Server.Flags().StringVar(&c.Kafka.GSSAPI.ServiceName, "gssapi-servicename", "kafka", "ServiceName")
-	Server.Flags().StringVar(&c.Kafka.GSSAPI.Username, "gssapi-username", "kafka", "Username")
-	Server.Flags().StringVar(&c.Kafka.GSSAPI.Realm, "gssapi-realm", "", "Realm")
-	Server.Flags().StringVar(&c.Kafka.GSSAPI.KerberosConfigPath, "gssapi-krb5", "/etc/krb5.conf", "krb5.conf file path, default: /etc/krb5.conf")
-	Server.Flags().StringVar(&c.Kafka.GSSAPI.KeyTabPath, "gssapi-keytab", "", "KeyTabPath")
+	Server.Flags().StringVar(&c.Kafka.SASL.GSSAPI.AuthType, "gssapi-auth-type", config.KRB5_KEYTAB_AUTH, "GSSAPI auth type: KEYTAB or USER")
+	Server.Flags().StringVar(&c.Kafka.SASL.GSSAPI.ServiceName, "gssapi-servicename", "kafka", "ServiceName")
+	Server.Flags().StringVar(&c.Kafka.SASL.GSSAPI.Username, "gssapi-username", "kafka", "Username")
+	Server.Flags().StringVar(&c.Kafka.SASL.GSSAPI.Password, "gssapi-password", "", "Password for auth type USER")
+	Server.Flags().StringVar(&c.Kafka.SASL.GSSAPI.Realm, "gssapi-realm", "", "Realm")
+	Server.Flags().StringVar(&c.Kafka.SASL.GSSAPI.KerberosConfigPath, "gssapi-krb5", "/etc/krb5.conf", "krb5.conf file path, default: /etc/krb5.conf")
+	Server.Flags().StringVar(&c.Kafka.SASL.GSSAPI.KeyTabPath, "gssapi-keytab", "", "krb5.keytab file location")
+	Server.Flags().BoolVar(&c.Kafka.SASL.GSSAPI.DisablePAFXFAST, "gssapi-disable-pa-fx-fast", false, "Used to configure the client to not use PA_FX_FAST.")
+	Server.Flags().StringToStringVar(&c.Kafka.SASL.GSSAPI.SPNHostsMapping, "gssapi-spn-host-mapping", map[string]string{}, "Mapping of Kafka servers address to SPN hosts")
 
 	// SASL by Proxy plugin
 	Server.Flags().BoolVar(&c.Kafka.SASL.Plugin.Enable, "sasl-plugin-enable", false, "Use plugin for SASL authentication")

config/config.go

@@ -11,7 +11,11 @@ import (
 	"github.com/pkg/errors"
 )
 
-const defaultClientID = "kafka-proxy"
+const (
+	defaultClientID  = "kafka-proxy"
+	KRB5_USER_AUTH   = "USER"
+	KRB5_KEYTAB_AUTH = "KEYTAB"
+)
 
 var (
 	// Version is the current version of the app, generated at build time
@@ -30,6 +34,18 @@ type DialAddressMapping struct {
 	DestinationAddress string
 }
 
+type GSSAPIConfig struct {
+	AuthType           string
+	KeyTabPath         string
+	KerberosConfigPath string
+	ServiceName        string
+	Username           string
+	Password           string
+	Realm              string
+	DisablePAFXFAST    bool
+	SPNHostsMapping    map[string]string
+}
+
 type Config struct {
 	Http struct {
 		ListenAddress string
@@ -144,14 +160,7 @@ type Config struct {
 				LogLevel   string
 				Timeout    time.Duration
 			}
-		}
-		GSSAPI struct {
-			Enable             bool
-			KeyTabPath         string
-			KerberosConfigPath string
-			ServiceName        string
-			Username           string
-			Realm              string
+			GSSAPI GSSAPIConfig
 		}
 		Producer struct {
 			Acks0Disabled bool
@@ -288,8 +297,32 @@ func (c *Config) Validate() error {
 				return errors.New("Mechanism OAUTHBEARER is required when Kafka.SASL.Plugin.Enable is enabled")
 			}
 		} else {
-			if c.Kafka.SASL.Username == "" || c.Kafka.SASL.Password == "" {
-				return errors.New("SASL.Username and SASL.Password are required when SASL is enabled and plugin is not used")
+			if c.Kafka.SASL.Method == "GSSAPI" {
+				switch c.Kafka.SASL.GSSAPI.AuthType {
+				case KRB5_USER_AUTH:
+					if c.Kafka.SASL.GSSAPI.Password == "" {
+						return errors.New("GSSAPI.Password is required for GSSAPI.AuthType USER")
+					}
+				case KRB5_KEYTAB_AUTH:
+					if c.Kafka.SASL.GSSAPI.KeyTabPath == "" {
+						return errors.New("GSSAPI.KeyTabPath is required for GSSAPI.AuthType KEYTAB")
+					}
+				default:
+					return errors.Errorf("Unsupported GSSAPI.AuthType %s", c.Kafka.SASL.GSSAPI.AuthType)
+				}
+				if c.Kafka.SASL.GSSAPI.KerberosConfigPath == "" {
+					return errors.New("GSSAPI KerberosConfigPath must not be empty")
+				}
+				if c.Kafka.SASL.GSSAPI.Username == "" {
+					return errors.New("GSSAPI Username must not be empty")
+				}
+				if c.Kafka.SASL.GSSAPI.Realm == "" {
+					return errors.New("GSSAPI Realm must not be empty")
+				}
+			} else {
+				if c.Kafka.SASL.Username == "" || c.Kafka.SASL.Password == "" {
+					return errors.New("SASL.Username and SASL.Password are required when SASL is enabled and plugin is not used")
+				}
 			}
 		}
 	} else {

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.1
 	github.com/sirupsen/logrus v1.6.0
-	github.com/spf13/cobra v0.0.1
+	github.com/spf13/cobra v1.6.1
 	github.com/spf13/viper v1.0.2
 	github.com/stretchr/testify v1.8.0
 	github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c
@@ -42,7 +42,7 @@ require (
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
 	github.com/hashicorp/hcl v0.0.0-20180404174102-ef8a98b0bbce // indirect
 	github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d // indirect
-	github.com/inconshreveable/mousetrap v1.0.0 // indirect
+	github.com/inconshreveable/mousetrap v1.0.1 // indirect
 	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
 	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
@@ -59,7 +59,7 @@ require (
 	github.com/spf13/afero v1.1.1 // indirect
 	github.com/spf13/cast v1.2.0 // indirect
 	github.com/spf13/jwalterweatherman v0.0.0-20180109140146-7c0cea34c8ec // indirect
-	github.com/spf13/pflag v1.0.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/xdg/stringprep v1.0.0 // indirect
 	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa // indirect
 	golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f // indirect

go.sum

@@ -19,6 +19,7 @@ github.com/cenkalti/backoff v1.1.0 h1:QnvVp8ikKCDWOsFheytRCoYWYPO/ObCTBGxT19Hc+y
 github.com/cenkalti/backoff v1.1.0/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -77,8 +78,8 @@ github.com/hashicorp/hcl v0.0.0-20180404174102-ef8a98b0bbce h1:xdsDDbiBDQTKASoGE
 github.com/hashicorp/hcl v0.0.0-20180404174102-ef8a98b0bbce/go.mod h1:oZtUIOe8dh44I2q6ScRibXws4Ajl+d+nod3AaR9vL5w=
 github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d h1:kJCB4vdITiW1eC1vq2e6IsrXKrZit1bv/TDYFGMp4BQ=
 github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM=
-github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
-github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
+github.com/inconshreveable/mousetrap v1.0.1 h1:U3uMjPSQEBMNp1lFxmllqCPM6P5u/Xq7Pgzkat/bFNc=
+github.com/inconshreveable/mousetrap v1.0.1/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
 github.com/jcmturner/aescts/v2 v2.0.0/go.mod h1:AiaICIRyfYg35RUkr8yESTqvSy7csK90qZ5xfvvsoNs=
 github.com/jcmturner/dnsutils/v2 v2.0.0 h1:lltnkeZGL0wILNvrNiVCR6Ro5PGU/SeBvVO/8c/iPbo=
@@ -148,6 +149,7 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
 github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/procfs v0.6.0 h1:mxy4L2jP6qMonqmq+aTtOx1ifVWUgG/TAmntgbh3xv4=
 github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0 h1:UBcNElsrwanuuMsnGSlYmtmgbb23qDR5dG+6X6Oo89I=
@@ -156,12 +158,12 @@ github.com/spf13/afero v1.1.1 h1:Lt3ihYMlE+lreX1GS4Qw4ZsNpYQLxIXKBTEOXm3nt6I=
 github.com/spf13/afero v1.1.1/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/cast v1.2.0 h1:HHl1DSRbEQN2i8tJmtS6ViPyHx35+p51amrdsiTCrkg=
 github.com/spf13/cast v1.2.0/go.mod h1:r2rcYCSwa1IExKTDiTfzaxqT2FNHs8hODu4LnUfgKEg=
-github.com/spf13/cobra v0.0.1 h1:zZh3X5aZbdnoj+4XkaBxKfhO4ot82icYdhhREIAXIj8=
-github.com/spf13/cobra v0.0.1/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
+github.com/spf13/cobra v1.6.1 h1:o94oiPyS4KD1mPy2fmcYYHHfCxLqYjJOhGsCHFZtEzA=
+github.com/spf13/cobra v1.6.1/go.mod h1:IOw/AERYS7UzyrGinqmz6HLUo219MORXGxhbaJUqzrY=
 github.com/spf13/jwalterweatherman v0.0.0-20180109140146-7c0cea34c8ec h1:2ZXvIUGghLpdTVHR1UfvfrzoVlZaE/yOWC5LueIHZig=
 github.com/spf13/jwalterweatherman v0.0.0-20180109140146-7c0cea34c8ec/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
-github.com/spf13/pflag v1.0.0 h1:oaPbdDe/x0UncahuwiPxW1GYJyilRAdsPnq3e1yaPcI=
-github.com/spf13/pflag v1.0.0/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
+github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.0.2 h1:Ncr3ZIuJn322w2k1qmzXDnkLAdQMlJqBa9kfAH+irso=
 github.com/spf13/viper v1.0.2/go.mod h1:A8kyI5cUJhb8N+3pkfONlcEcZbueH6nhAm0Fq7SrnBM=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=

proxy/auth.go

@@ -21,7 +21,7 @@ type AuthClient struct {
 	tokenProvider apis.TokenProvider
 }
 
-//TODO: reset deadlines after method - ok
+// TODO: reset deadlines after method - ok
 func (b *AuthClient) sendAndReceiveGatewayAuth(conn DeadlineReaderWriter) error {
 	//TODO: timeout
 	//	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(p.timeout)*time.Second)
@@ -77,7 +77,7 @@ type AuthServer struct {
 	tokenInfo apis.TokenInfo
 }
 
-//TODO: reset deadlines after method - ok
+// TODO: reset deadlines after method - ok
 func (b *AuthServer) receiveAndSendGatewayAuth(conn DeadlineReaderWriter) error {
 	err := conn.SetDeadline(time.Now().Add(b.timeout))
 	if err != nil {

proxy/client.go

@@ -28,7 +28,7 @@ type Client struct {
 	// Kafka Net configuration
 	config *config.Config
 
-	// Config of Proxy request-response processor (instance p)
+	// config of Proxy request-response processor (instance p)
 	processorConfig ProcessorConfig
 
 	dialer         Dialer
@@ -39,7 +39,6 @@ type Client struct {
 
 	saslAuthByProxy SASLAuthByProxy
 	authClient      *AuthClient
-	gssapiKerberosAuth *GSSAPIKerberosAuth
 
 	dialAddressMapping map[string]config.DialAddressMapping
 
@@ -118,32 +117,24 @@ func NewClient(conns *ConnSet, c *config.Config, netAddressMappingFunc config.Ne
 				password:     c.Kafka.SASL.Password,
 				mechanism:    c.Kafka.SASL.Method,
 			}
+		} else if c.Kafka.SASL.Method == SASLSGSSAPI {
+			saslAuthByProxy = &SASLGSSAPIAuth{
+				writeTimeout: c.Kafka.WriteTimeout,
+				readTimeout:  c.Kafka.ReadTimeout,
+				gssapiConfig: &c.Kafka.SASL.GSSAPI,
+			}
 		} else {
 			return nil, errors.Errorf("SASL Mechanism not valid '%s'", c.Kafka.SASL.Method)
 		}
 	}
 
-	var gssapiKerberosAuth *GSSAPIKerberosAuth
-	if c.Kafka.GSSAPI.Enable {
-		gssapiKerberosAuth = &GSSAPIKerberosAuth{
-			Config: &GSSAPIConfig{
-				KeyTabPath:         c.Kafka.GSSAPI.KeyTabPath,
-				KerberosConfigPath: c.Kafka.GSSAPI.KerberosConfigPath,
-				ServiceName:        c.Kafka.GSSAPI.ServiceName,
-				Username:           c.Kafka.GSSAPI.Username,
-				Realm:              c.Kafka.GSSAPI.Realm,
-			},
-		}
-	}
-
 	dialAddressMapping, err := getAddressToDialAddressMapping(c)
 	if err != nil {
 		return nil, err
 	}
 
 	return &Client{conns: conns, config: c, dialer: dialer, tcpConnOptions: tcpConnOptions, stopRun: make(chan struct{}, 1),
-		saslAuthByProxy:    saslAuthByProxy,
-		gssapiKerberosAuth: gssapiKerberosAuth,
+		saslAuthByProxy: saslAuthByProxy,
 		authClient: &AuthClient{
 			enabled:       c.Auth.Gateway.Client.Enable,
 			magic:         c.Auth.Gateway.Client.Magic,
@@ -339,19 +330,7 @@ func (c *Client) auth(conn net.Conn, brokerAddress string) error {
 		}
 	}
 	if c.config.Kafka.SASL.Enable {
-		err := c.saslAuthByProxy.sendAndReceiveSASLAuth(conn)
-		if err != nil {
-			_ = conn.Close()
-			return err
-		}
-		if err := conn.SetDeadline(time.Time{}); err != nil {
-			_ = conn.Close()
-			return err
-		}
-	}
-
-	if c.config.Kafka.GSSAPI.Enable {
-		err := c.gssapiKerberosAuth.sendAndReceiveGSSAPIAuth(conn, brokerAddress)
+		err := c.saslAuthByProxy.sendAndReceiveSASLAuth(conn, brokerAddress)
 		if err != nil {
 			_ = conn.Close()
 			return err

proxy/protocol/responses_test.go

@@ -2956,13 +2956,13 @@ func TestFindCoordinatorResponse(t *testing.T) {
 			modifier: testResponseModifier2,
 			modified: []string{"throttle_time_ms int32 0", "error_code int16 8", "error_message *string The broker is not available.", "coordinator struct", "node_id int32 0", "host string myhost", "port int32 34000", "[response_tagged_fields]", "response_tagged_fields tag 0 value 0x4711"},
 		},
-		{name: "v4, one coordinator" , apiVersion: int16(4),
+		{name: "v4, one coordinator", apiVersion: int16(4),
 			hexInput: "0000000002066b65792d32000000020a6c6f63616c686f7374000071a40000010000",
 			expected: []string{"throttle_time_ms int32 0", "[coordinators]", "coordinators struct", "key string key-2", "coordinator struct", "node_id int32 2", "host string localhost", "port int32 29092", "error_code int16 0", "error_message *string ", "[coordinators_tagged_fields]", "[response_tagged_fields]"},
 			modifier: testResponseModifier2,
 			modified: []string{"throttle_time_ms int32 0", "[coordinators]", "coordinators struct", "key string key-2", "coordinator struct", "node_id int32 2", "host string myhost2", "port int32 34002", "error_code int16 0", "error_message *string ", "[coordinators_tagged_fields]", "[response_tagged_fields]"},
 		},
-		{name: "v4, multiple coordinators" , apiVersion: int16(4),
+		{name: "v4, multiple coordinators", apiVersion: int16(4),
 			hexInput: "0000000004066b65792d31000000010a6c6f63616c686f737400004a9400000100066b65792d32000000020a6c6f63616c686f7374000071a400000101e724087461672034373131066b65792d33000000030a6c6f63616c686f7374000098b4000d0f4572726f7220636f64652031332e0000",
 			expected: []string{"throttle_time_ms int32 0", "[coordinators]", "coordinators struct", "key string key-1", "coordinator struct", "node_id int32 1", "host string localhost", "port int32 19092", "error_code int16 0", "error_message *string ", "[coordinators_tagged_fields]", "coordinators struct", "key string key-2", "coordinator struct", "node_id int32 2", "host string localhost", "port int32 29092", "error_code int16 0", "error_message *string ", "[coordinators_tagged_fields]", "coordinators_tagged_fields tag 4711 value 0x7461672034373131", "coordinators struct", "key string key-3", "coordinator struct", "node_id int32 3", "host string localhost", "port int32 39092", "error_code int16 13", "error_message *string Error code 13.", "[coordinators_tagged_fields]", "[response_tagged_fields]"},
 			modifier: testResponseModifier2,

proxy/protocol/schema.go

@@ -44,7 +44,6 @@ type schema struct {
 	fieldsByName map[string]*boundField
 }
 
-//
 type Mfield struct {
 	Name string
 	Ty   Schema