SASL in processing loop

Author: everesio

Makefile

@@ -8,7 +8,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.0.1"
+TAG           ?= "v0.0.2"
 
 PLATFORM      ?= $(shell uname -s)
 ifeq ($(PLATFORM), Darwin)

README.md

@@ -17,6 +17,9 @@ The dynamic local listeners feature can be disabled and an additional list of ex
 The Proxy can terminate TLS traffic and authenticate users using SASL/PLAIN. The credentials verification method
 is configurable and uses golang plugin system over RPC.
 
+The proxies can also authenticate each other using a pluggable method which is transparent to other Kafka servers and clients.
+Currently the Google ID Token for service accounts is implemented i.e. proxy client requests and sends service account JWT and proxy server receives and validates it against Google JWKS.
+
 Kafka API calls can be restricted to prevent some operations e.g. topic deletion or produce requests.
 
 
@@ -31,11 +34,11 @@ See:
 
    Linux
 
-        curl -Lso kafka-proxy https://github.com/grepplabs/kafka-proxy/releases/download/v0.0.1/linux.amd64.kafka-proxy
+        curl -Lso kafka-proxy https://github.com/grepplabs/kafka-proxy/releases/download/v0.0.2/linux.amd64.kafka-proxy
 
    macOS
 
-        curl -Lso kafka-proxy https://github.com/grepplabs/kafka-proxy/releases/download/v0.0.1/darwin.amd64.kafka-proxy
+        curl -Lso kafka-proxy https://github.com/grepplabs/kafka-proxy/releases/download/v0.0.2/darwin.amd64.kafka-proxy
 
 2. Make the kafka-proxy binary executable
 

proxy/processor.go

@@ -18,6 +18,10 @@ const (
 	defaultWriteTimeout       = 30 * time.Second
 	defaultReadTimeout        = 30 * time.Second
 	minOpenRequests           = 1
+
+	apiKeyUnset         = int16(-1) // not in protocol
+	apiKeySaslAuth      = int16(-2) // not in protocol
+	apiKeySaslHandshake = int16(17)
 )
 
 type ProcessorConfig struct {
@@ -108,8 +112,16 @@ func requestsLoop(dst DeadlineWriter, src DeadlineReader, openRequestsChannel ch
 	keyVersionBuf := make([]byte, 8) // Size => int32 + ApiKey => int16 + ApiVersion => int16
 
 	buf := make([]byte, bufSize)
-
+	lastApiKey := apiKeyUnset
+nextRequest:
 	for {
+		if lastApiKey == apiKeySaslHandshake {
+			lastApiKey = apiKeySaslAuth
+			if readErr, err = copySaslAuthRequest(dst, src, timeout, buf); err != nil {
+				return readErr, err
+			}
+			continue nextRequest
+		}
 		// logrus.Println("Await Kafka request")
 
 		// waiting for first bytes or EOF - reset deadlines
@@ -124,7 +136,7 @@ func requestsLoop(dst DeadlineWriter, src DeadlineReader, openRequestsChannel ch
 		if err = protocol.Decode(keyVersionBuf, requestKeyVersion); err != nil {
 			return true, err
 		}
-		// logrus.Printf("Kafka request length %v, key %v, version %v", requestKeyVersion.Length, requestKeyVersion.ApiKey, requestKeyVersion.ApiVersion)
+		//logrus.Printf("Kafka request length %v, key %v, version %v", requestKeyVersion.Length, requestKeyVersion.ApiKey, requestKeyVersion.ApiVersion)
 
 		proxyRequestsTotal.WithLabelValues(brokerAddress, strconv.Itoa(int(requestKeyVersion.ApiKey)), strconv.Itoa(int(requestKeyVersion.ApiVersion))).Inc()
 		proxyRequestsBytes.WithLabelValues(brokerAddress).Add(float64(requestKeyVersion.Length + 4))
@@ -156,15 +168,26 @@ func requestsLoop(dst DeadlineWriter, src DeadlineReader, openRequestsChannel ch
 		if readErr, err = myCopyN(dst, src, int64(requestKeyVersion.Length-4), buf); err != nil {
 			return readErr, err
 		}
+
+		lastApiKey = requestKeyVersion.ApiKey
 	}
 }
 
 func responsesLoop(dst DeadlineWriter, src DeadlineReader, openRequestsChannel <-chan protocol.RequestKeyVersion, netAddressMappingFunc config.NetAddressMappingFunc, bufSize int, timeout time.Duration, brokerAddress string) (readErr bool, err error) {
 	responseHeaderBuf := make([]byte, 8) // Size => int32, CorrelationId => int32
 
 	buf := make([]byte, bufSize)
+	lastApiKey := apiKeyUnset
 
+nextResponse:
 	for {
+		if lastApiKey == apiKeySaslHandshake {
+			lastApiKey = apiKeySaslAuth
+			if readErr, err = copySaslAuthResponse(dst, src, timeout); err != nil {
+				return readErr, err
+			}
+			continue nextResponse
+		}
 		//logrus.Println("Await Kafka response")
 
 		// waiting for first bytes or EOF - reset deadlines
@@ -197,7 +220,7 @@ func responsesLoop(dst DeadlineWriter, src DeadlineReader, openRequestsChannel <
 		if err != nil {
 			return true, err
 		}
-		// logrus.Printf("Kafka response length %v, key %v, version %v", requestKeyVersion.Length, requestKeyVersion.ApiKey, requestKeyVersion.ApiVersion)
+
 		responseModifier, err := protocol.GetResponseModifier(requestKeyVersion.ApiKey, requestKeyVersion.ApiVersion, netAddressMappingFunc)
 		if err != nil {
 			return true, err
@@ -235,6 +258,7 @@ func responsesLoop(dst DeadlineWriter, src DeadlineReader, openRequestsChannel <
 				return readErr, err
 			}
 		}
+		lastApiKey = requestKeyVersion.ApiKey
 	}
 }
 

proxy/sasl_forward.go

@@ -0,0 +1,66 @@
+package proxy
+
+import (
+	"encoding/binary"
+	"fmt"
+	"github.com/grepplabs/kafka-proxy/proxy/protocol"
+	"io"
+	"time"
+)
+
+func copySaslAuthRequest(dst DeadlineWriter, src DeadlineReader, timeout time.Duration, buf []byte) (readErr bool, err error) {
+	requestDeadline := time.Now().Add(timeout)
+	err = dst.SetWriteDeadline(requestDeadline)
+	if err != nil {
+		return false, err
+	}
+	err = src.SetReadDeadline(requestDeadline)
+	if err != nil {
+		return true, err
+	}
+
+	sizeBuf := make([]byte, 4) // Size => int32
+	if _, err = io.ReadFull(src, sizeBuf); err != nil {
+		return true, err
+	}
+
+	length := binary.BigEndian.Uint32(sizeBuf)
+	if int32(length) > protocol.MaxRequestSize {
+		return true, protocol.PacketDecodingError{Info: fmt.Sprintf("auth message of length %d too large", length)}
+	}
+	//logrus.Printf("SASL auth request length %v", length)
+
+	// write - send to broker
+	if _, err = dst.Write(sizeBuf); err != nil {
+		return false, err
+	}
+	if readErr, err = myCopyN(dst, src, int64(length), buf); err != nil {
+		return readErr, err
+	}
+	return false, nil
+}
+
+func copySaslAuthResponse(dst DeadlineWriter, src DeadlineReader, timeout time.Duration) (readErr bool, err error) {
+	//logrus.Printf("SASL auth response")
+
+	responseDeadline := time.Now().Add(timeout)
+	err = dst.SetWriteDeadline(responseDeadline)
+	if err != nil {
+		return false, err
+	}
+	err = src.SetReadDeadline(responseDeadline)
+	if err != nil {
+		return true, err
+	}
+
+	header := make([]byte, 4)
+	_, err = io.ReadFull(src, header)
+	if err != nil {
+		return true, err
+	}
+
+	if _, err = dst.Write(header); err != nil {
+		return false, err
+	}
+	return false, nil
+}