hashicorp/go-plugin

Author: everesio

.gitignore

@@ -1,6 +1,7 @@
 # Build
 build/
 /kafka-proxy
+testcerts/
 
 # Intellij
 .idea/
@@ -57,4 +58,4 @@ Session.vim
 .netrwhist
 *~
 # Auto-generated tag files
-tags
+tags

Gopkg.lock

@@ -68,5 +68,11 @@ build.docker-build.osx:
     docker rm $$buildContainer ;\
     docker rmi $$buildContainerName ;\
 
+protoc.auth:
+	protoc -I plugin/auth/proto/ plugin/auth/proto/auth.proto --go_out=plugins=grpc:plugin/auth/proto/
+
+plugin.auth-user:
+	CGO_ENABLED=0 go build -o build/auth-user $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-auth-user/main.go
+
 clean:
-	@rm -rf build
+	@rm -rf build

Makefile

@@ -14,8 +14,8 @@ responses received from the brokers are replaced by local counterparts.
 For discovered brokers (not configured as the boostrap servers), local listeners are started on random ports.
 The dynamic local listeners feature can be disabled and an additional list of external server mappings can be provided.
 
-The Proxy can terminate TLS traffic and authenticate users locally using SASL/PLAIN. The credentials check method 
-is configurable by providing dynamically linked shared object library.   
+The Proxy can terminate TLS traffic and authenticate users locally using SASL/PLAIN. The credentials verification method
+is configurable and uses golang plugin system over RPC.
 
 Kafka API calls can be restricted to prevent some operations e.g. topic deletion.
 
@@ -40,11 +40,23 @@ See:
                              --tls-enable --tls-insecure-skip-verify \
                              --sasl-enable -sasl-username myuser --sasl-password mysecret
 
-    build/kafka-proxy server --proxy-listener-key-file "server-key.pem"  \
+    make clean build plugin.auth-user && build/kafka-proxy server \
+                             --proxy-listener-auth-enable \
+                             --proxy-listener-auth-command build/auth-user \
+                             --proxy-listener-auth-param "--username=my-test-user" \
+                             --proxy-listener-auth-param "--password=my-test-password" \
+                             --dynamic-listeners-disable \
+                             --forbidden-api-keys 20 \
+                             --bootstrap-server-mapping "192.168.99.100:32400,127.0.0.1:32401"
+
+    make clean build plugin.auth-user && build/kafka-proxy server --proxy-listener-key-file "server-key.pem"  \
                              --proxy-listener-cert-file "server-cert.pem" \
                              --proxy-listener-ca-chain-cert-file "ca.pem" \
                              --proxy-listener-tls-enable \
                              --proxy-listener-auth-enable \
+                             --proxy-listener-auth-command build/auth-user \
+                             --proxy-listener-auth-param "--username=my-test-user" \
+                             --proxy-listener-auth-param "--password=my-test-password" \                             
                              --dynamic-listeners-disable \
                              --forbidden-api-keys 20 \
                              --bootstrap-server-mapping "192.168.99.100:32400,127.0.0.1:32401" \
@@ -65,7 +77,7 @@ See:
   3. counter: proxy_connections_total {broker}
   4. counter: proxy_requests_bytes {broker}
   5. counter: proxy_responses_bytes {broker}
-* [ ] Plugable local authentication
+* [X] Pluggable local authentication
 * [ ] Deploying Kafka Proxy as a sidecar container
 * [ ] Performance tests and tuning
 * [ ] Socket buffer sizing e.g. SO_RCVBUF = 32768, SO_SNDBUF = 131072

README.md

@@ -13,9 +13,14 @@ import (
 	"net/http"
 	_ "net/http/pprof"
 	"os"
+	"os/exec"
 	"os/signal"
 	"syscall"
 	"time"
+
+	"errors"
+	"github.com/grepplabs/kafka-proxy/plugin/auth/shared"
+	"github.com/hashicorp/go-plugin"
 )
 
 var (
@@ -69,7 +74,10 @@ func init() {
 	Server.Flags().StringVar(&c.Proxy.TLS.ListenerKeyPassword, "proxy-listener-key-password", "", "Password to decrypt rsa private key")
 	Server.Flags().StringVar(&c.Proxy.TLS.CAChainCertFile, "proxy-listener-ca-chain-cert-file", "", "PEM encoded CA's certificate file")
 
+	// authentication plugin
 	Server.Flags().BoolVar(&c.Proxy.Auth.Enable, "proxy-listener-auth-enable", false, "Enable SASL/PLAIN listener authentication")
+	Server.Flags().StringVar(&c.Proxy.Auth.Command, "proxy-listener-auth-command", "", "Path to authentication plugin binary")
+	Server.Flags().StringSliceVar(&c.Proxy.Auth.Parameters, "proxy-listener-auth-param", []string{}, "Authentication plugin parameter")
 
 	// kafka
 	Server.Flags().StringVar(&c.Kafka.ClientID, "kafka-client-id", "kafka-proxy", "An optional identifier to track the source of requests")
@@ -116,6 +124,34 @@ func init() {
 func Run(_ *cobra.Command, _ []string) {
 	logrus.Infof("Starting kafka-proxy version %s", config.Version)
 
+	var passwordAuthenticator shared.PasswordAuthenticator
+	if c.Proxy.Auth.Enable {
+		client := plugin.NewClient(&plugin.ClientConfig{
+			HandshakeConfig: shared.Handshake,
+			Plugins:         shared.PluginMap,
+			SyncStdout:      os.Stdout,
+			SyncStderr:      os.Stderr,
+			Cmd:             exec.Command(c.Proxy.Auth.Command, c.Proxy.Auth.Parameters...),
+			AllowedProtocols: []plugin.Protocol{
+				plugin.ProtocolNetRPC, plugin.ProtocolGRPC},
+		})
+		defer client.Kill()
+		rpcClient, err := client.Client()
+		if err != nil {
+			logrus.Fatal(err)
+		}
+		raw, err := rpcClient.Dispense("passwordAuthenticator")
+		if err != nil {
+			logrus.Fatal(err)
+		}
+		var ok bool
+		passwordAuthenticator, ok = raw.(shared.PasswordAuthenticator)
+		if !ok {
+			logrus.Fatal(errors.New("unsupported plugin type"))
+		}
+	}
+	_ = passwordAuthenticator
+
 	var g group.Group
 	{
 		// All active connections are stored in this variable.
@@ -129,7 +165,7 @@ func Run(_ *cobra.Command, _ []string) {
 		if err != nil {
 			logrus.Fatal(err)
 		}
-		proxyClient, err := proxy.NewClient(connset, c, listeners.GetNetAddressMapping)
+		proxyClient, err := proxy.NewClient(connset, c, listeners.GetNetAddressMapping, passwordAuthenticator)
 		if err != nil {
 			logrus.Fatal(err)
 		}
@@ -179,6 +215,7 @@ func Run(_ *cobra.Command, _ []string) {
 			debugListener.Close()
 		})
 	}
+
 	err := g.Run()
 	logrus.Info("Exit", err)
 }

cmd/kafka-proxy/server.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"flag"
+	"github.com/grepplabs/kafka-proxy/plugin/auth/shared"
+	"github.com/hashicorp/go-plugin"
+	"github.com/sirupsen/logrus"
+	"os"
+)
+
+type PasswordAuthenticator struct {
+	username string
+	password string
+}
+
+func (pa PasswordAuthenticator) Authenticate(username, password string) (bool, int32, error) {
+	// logrus.Printf("Authenticate request for %s:%s,expected %s:%s ", username, password, pa.username, pa.password)
+	return username == pa.username && password == pa.password, 0, nil
+}
+
+type PluginMeta struct {
+	flagUsername string
+	flagPassword string
+}
+
+func (f *PluginMeta) FlagSet() *flag.FlagSet {
+	fs := flag.NewFlagSet("auth plugin settings", flag.ContinueOnError)
+	fs.StringVar(&f.flagUsername, "username", "", "")
+	fs.StringVar(&f.flagPassword, "password", "", "")
+	return fs
+}
+
+func main() {
+
+	pluginMeta := &PluginMeta{}
+	flags := pluginMeta.FlagSet()
+	flags.Parse(os.Args[1:])
+
+	if pluginMeta.flagUsername == "" || pluginMeta.flagPassword == "" {
+		logrus.Errorf("parameters username and password are required")
+		os.Exit(1)
+	}
+
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: shared.Handshake,
+		Plugins: map[string]plugin.Plugin{
+			"passwordAuthenticator": &shared.PasswordAuthenticatorPlugin{Impl: &PasswordAuthenticator{
+				username: pluginMeta.flagUsername,
+				password: pluginMeta.flagPassword,
+			}},
+		},
+		// A non-nil value here enables gRPC serving for this plugin...
+		GRPCServer: plugin.DefaultGRPCServer,
+	})
+}

cmd/plugin-auth-user/main.go

@@ -57,7 +57,9 @@ type Config struct {
 			CAChainCertFile     string
 		}
 		Auth struct {
-			Enable bool
+			Enable     bool
+			Command    string
+			Parameters []string
 		}
 	}
 	Kafka struct {
@@ -205,5 +207,8 @@ func (c *Config) Validate() error {
 	if c.Proxy.TLS.Enable && (c.Proxy.TLS.ListenerKeyFile == "" || c.Proxy.TLS.ListenerCertFile == "") {
 		return errors.New("ListenerKeyFile and ListenerCertFile are required when Proxy TLS is enabled")
 	}
+	if c.Proxy.Auth.Enable && c.Proxy.Auth.Command == "" {
+		return errors.New("Auth.Command is required when Proxy.Auth is enabled")
+	}
 	return nil
 }