Built-in component registry

Author: everesio

.travis.yml

@@ -24,7 +24,6 @@ notifications:
 
 script:
   - go build .
-  - go build -o google-id-info cmd/plugin-googleid-info/main.go
   - export TAG=`if [[ $TRAVIS_PULL_REQUEST == "false" ]] && [[ $TRAVIS_BRANCH == "master" ]]; then echo "latest"; else echo "${TRAVIS_PULL_REQUEST_BRANCH:-${TRAVIS_BRANCH}}"; fi`
   - docker build -t $REPO:$TAG -f Dockerfile .
 

Dockerfile

@@ -3,7 +3,6 @@ FROM alpine:3.7
 RUN apk add --no-cache ca-certificates
 
 ADD kafka-proxy /kafka-proxy
-ADD google-id-info /google-id-info
 
 ENTRYPOINT ["/kafka-proxy"]
 CMD ["--help"]

Makefile

@@ -38,15 +38,12 @@ build: build/$(BINARY)
 
 build/$(BINARY): $(SOURCES)
 	CGO_ENABLED=0 go build -o build/$(BINARY) $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" .
-	CGO_ENABLED=0 go build -o build/google-id-info $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-googleid-info/main.go
 
 build/linux/$(BINARY): $(SOURCES)
 	GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o build/linux/$(BINARY) $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" .
-	GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o build/linux/google-id-info $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-googleid-info/main.go
 
 build/osx/$(BINARY): $(SOURCES)
 	GOOS=darwin GOARCH=amd64 CGO_ENABLED=0 go build -o build/osx/$(BINARY) $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" .
-	GOOS=darwin GOARCH=amd64 CGO_ENABLED=0 go build -o build/linux/google-id-info $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-googleid-info/main.go
 
 build.docker-build: $(BUILD_DOCKER_BUILD)
 
@@ -58,7 +55,6 @@ build.docker-build.linux:
     echo "containerId: $$buildContainer" ;\
     mkdir -p build ;\
     docker cp $$buildContainer:/go/src/github.com/grepplabs/kafka-proxy/build/linux/${BINARY} build/${BINARY} ;\
-    docker cp $$buildContainer:/go/src/github.com/grepplabs/kafka-proxy/build/linux/google-id-info build/google-id-info ;\
     docker rm $$buildContainer ;\
     docker rmi $$buildContainerName ;\
 
@@ -70,7 +66,6 @@ build.docker-build.osx:
     echo "containerId: $$buildContainer" ;\
     mkdir -p build ;\
     docker cp $$buildContainer:/go/src/github.com/grepplabs/kafka-proxy/build/osx/${BINARY} build/${BINARY} ;\
-    docker cp $$buildContainer:/go/src/github.com/grepplabs/kafka-proxy/build/linux/google-id-info build/google-id-info ;\
     docker rm $$buildContainer ;\
     docker rmi $$buildContainerName ;\
 

README.md

@@ -280,7 +280,7 @@ spec:
 * [X] Pluggable authentication between client kafka-proxy and broker kafka-proxy a.k.a kafka-gateway
   1. additional handshake - protocol: magic, method, data
   2. google-id method
-* [ ] Registry for built in plugins (avoid grpc communication)
+* [X] Registry for built-in plugins
 * [ ] Client cert check
 * [ ] TLS server parameters like CipherSuites etc. - see ory/graceful/blob/master/http_defaults.go
 * [ ] Performance tests and tuning

cmd/kafka-proxy/server.go

@@ -25,6 +25,11 @@ import (
 	localauth "github.com/grepplabs/kafka-proxy/plugin/local-auth/shared"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-plugin"
+
+	"github.com/grepplabs/kafka-proxy/pkg/registry"
+	// built-in plugins
+	_ "github.com/grepplabs/kafka-proxy/pkg/libs/googleid-info"
+	_ "github.com/grepplabs/kafka-proxy/pkg/libs/googleid-provider"
 )
 
 var (
@@ -148,61 +153,89 @@ func Run(_ *cobra.Command, _ []string) {
 
 	var passwordAuthenticator apis.PasswordAuthenticator
 	if c.Auth.Local.Enable {
-		client := NewLocalAuthPluginClient()
-		defer client.Kill()
+		var err error
+		factory, ok := registry.GetComponent(new(apis.PasswordAuthenticatorFactory), c.Auth.Local.Command).(apis.PasswordAuthenticatorFactory)
+		if ok {
+			logrus.Infof("Using built-in PasswordAuthenticator")
+			passwordAuthenticator, err = factory.New(c.Auth.Local.Parameters)
+			if err != nil {
+				logrus.Fatal(err)
+			}
+		} else {
+			client := NewPluginClient(localauth.Handshake, localauth.PluginMap, c.Auth.Local.LogLevel, c.Auth.Local.Command, c.Auth.Local.Parameters)
+			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.(apis.PasswordAuthenticator)
-		if !ok {
-			logrus.Fatal(errors.New("unsupported PasswordAuthenticator plugin type"))
+			rpcClient, err := client.Client()
+			if err != nil {
+				logrus.Fatal(err)
+			}
+			raw, err := rpcClient.Dispense("passwordAuthenticator")
+			if err != nil {
+				logrus.Fatal(err)
+			}
+			passwordAuthenticator, ok = raw.(apis.PasswordAuthenticator)
+			if !ok {
+				logrus.Fatal(errors.New("unsupported PasswordAuthenticator plugin type"))
+			}
 		}
 	}
 
 	var tokenProvider apis.TokenProvider
 	if c.Auth.Gateway.Client.Enable {
-		client := NewGatewayClientPluginClient()
-		defer client.Kill()
+		var err error
+		factory, ok := registry.GetComponent(new(apis.TokenProviderFactory), c.Auth.Gateway.Client.Command).(apis.TokenProviderFactory)
+		if ok {
+			logrus.Infof("Using built-in TokenProvider")
+			tokenProvider, err = factory.New(c.Auth.Gateway.Client.Parameters)
+			if err != nil {
+				logrus.Fatal(err)
+			}
+		} else {
+			client := NewPluginClient(gatewayclient.Handshake, gatewayclient.PluginMap, c.Auth.Gateway.Client.LogLevel, c.Auth.Gateway.Client.Command, c.Auth.Gateway.Client.Parameters)
+			defer client.Kill()
 
-		rpcClient, err := client.Client()
-		if err != nil {
-			logrus.Fatal(err)
-		}
-		raw, err := rpcClient.Dispense("tokenProvider")
-		if err != nil {
-			logrus.Fatal(err)
-		}
-		var ok bool
-		tokenProvider, ok = raw.(apis.TokenProvider)
-		if !ok {
-			logrus.Fatal(errors.New("unsupported TokenProvider plugin type"))
+			rpcClient, err := client.Client()
+			if err != nil {
+				logrus.Fatal(err)
+			}
+			raw, err := rpcClient.Dispense("tokenProvider")
+			if err != nil {
+				logrus.Fatal(err)
+			}
+			tokenProvider, ok = raw.(apis.TokenProvider)
+			if !ok {
+				logrus.Fatal(errors.New("unsupported TokenProvider plugin type"))
+			}
 		}
 	}
 
 	var tokenInfo apis.TokenInfo
 	if c.Auth.Gateway.Server.Enable {
-		client := NewGatewayServerPluginClient()
-		defer client.Kill()
+		var err error
+		factory, ok := registry.GetComponent(new(apis.TokenInfoFactory), c.Auth.Gateway.Server.Command).(apis.TokenInfoFactory)
+		if ok {
+			logrus.Infof("Using built-in TokenInfo")
+
+			tokenInfo, err = factory.New(c.Auth.Gateway.Server.Parameters)
+			if err != nil {
+				logrus.Fatal(err)
+			}
+		} else {
+			client := NewPluginClient(gatewayserver.Handshake, gatewayserver.PluginMap, c.Auth.Gateway.Server.LogLevel, c.Auth.Gateway.Server.Command, c.Auth.Gateway.Server.Parameters)
+			defer client.Kill()
 
-		rpcClient, err := client.Client()
-		if err != nil {
-			logrus.Fatal(err)
-		}
-		raw, err := rpcClient.Dispense("tokenInfo")
-		if err != nil {
-			logrus.Fatal(err)
-		}
-		var ok bool
-		tokenInfo, ok = raw.(apis.TokenInfo)
-		if !ok {
-			logrus.Fatal(errors.New("unsupported TokenInfo plugin type"))
+			rpcClient, err := client.Client()
+			if err != nil {
+				logrus.Fatal(err)
+			}
+			raw, err := rpcClient.Dispense("tokenInfo")
+			if err != nil {
+				logrus.Fatal(err)
+			}
+			tokenInfo, ok = raw.(apis.TokenInfo)
+			if !ok {
+				logrus.Fatal(errors.New("unsupported TokenInfo plugin type"))
+			}
 		}
 	}
 
@@ -318,16 +351,6 @@ func SetLogger() {
 	logrus.SetLevel(level)
 }
 
-func NewGatewayClientPluginClient() *plugin.Client {
-	return NewPluginClient(gatewayclient.Handshake, gatewayclient.PluginMap, c.Auth.Gateway.Client.LogLevel, c.Auth.Gateway.Client.Command, c.Auth.Gateway.Client.Parameters)
-}
-func NewGatewayServerPluginClient() *plugin.Client {
-	return NewPluginClient(gatewayserver.Handshake, gatewayserver.PluginMap, c.Auth.Gateway.Server.LogLevel, c.Auth.Gateway.Server.Command, c.Auth.Gateway.Server.Parameters)
-}
-func NewLocalAuthPluginClient() *plugin.Client {
-	return NewPluginClient(localauth.Handshake, localauth.PluginMap, c.Auth.Local.LogLevel, c.Auth.Local.Command, c.Auth.Local.Parameters)
-}
-
 func NewPluginClient(handshakeConfig plugin.HandshakeConfig, plugins map[string]plugin.Plugin, logLevel string, command string, params []string) *plugin.Client {
 	jsonFormat := false
 	if c.Log.Format == "json" {

cmd/plugin-googleid-info/main.go

@@ -1,66 +1,17 @@
 package main
 
 import (
-	"flag"
-	"fmt"
 	"github.com/grepplabs/kafka-proxy/pkg/libs/googleid-info"
 	"github.com/grepplabs/kafka-proxy/plugin/gateway-server/shared"
 	"github.com/hashicorp/go-plugin"
 	"github.com/sirupsen/logrus"
 	"os"
 )
 
-func (f *pluginMeta) flagSet() *flag.FlagSet {
-	fs := flag.NewFlagSet("google-id info settings", flag.ContinueOnError)
-	return fs
-}
-
-type pluginMeta struct {
-	timeout              int
-	certsRefreshInterval int
-	audience             arrayFlags
-	emailsRegex          arrayFlags
-}
-
-type arrayFlags []string
-
-func (i *arrayFlags) String() string {
-	return fmt.Sprintf("%v", *i)
-}
-
-func (i *arrayFlags) Set(value string) error {
-	*i = append(*i, value)
-	return nil
-}
-
-func (i *arrayFlags) asMap() map[string]struct{} {
-	result := make(map[string]struct{})
-	for _, elem := range *i {
-		result[elem] = struct{}{}
-	}
-	return result
-}
-
 func main() {
-	pluginMeta := &pluginMeta{}
-	fs := pluginMeta.flagSet()
-	fs.IntVar(&pluginMeta.timeout, "timeout", 10, "Request timeout in seconds")
-	fs.IntVar(&pluginMeta.certsRefreshInterval, "certs-refresh-interval", 60*60, "Certificates refresh interval in seconds")
-	fs.Var(&pluginMeta.audience, "audience", "The audience of a token")
-	fs.Var(&pluginMeta.emailsRegex, "email-regex", "Regex of the email claim")
-
-	fs.Parse(os.Args[1:])
-
-	opts := googleidinfo.TokenInfoOptions{
-		Timeout:              pluginMeta.timeout,
-		CertsRefreshInterval: pluginMeta.certsRefreshInterval,
-		Audience:             pluginMeta.audience,
-		EmailsRegex:          pluginMeta.emailsRegex,
-	}
-
-	tokenInfo, err := googleidinfo.NewTokenInfo(opts)
+	tokenInfo, err := new(googleidinfo.Factory).New(os.Args[1:])
 	if err != nil {
-		logrus.Errorf("cannot initialize googleid-info provider: %v", err)
+		logrus.Errorf("cannot initialize google-id-info provider: %v", err)
 		os.Exit(1)
 	}
 

cmd/plugin-googleid-provider/main.go

@@ -1,50 +1,17 @@
 package main
 
 import (
-	"flag"
 	"github.com/grepplabs/kafka-proxy/pkg/libs/googleid-provider"
 	"github.com/grepplabs/kafka-proxy/plugin/gateway-client/shared"
 	"github.com/hashicorp/go-plugin"
 	"github.com/sirupsen/logrus"
 	"os"
 )
 
-func (f *pluginMeta) flagSet() *flag.FlagSet {
-	fs := flag.NewFlagSet("google-id provider settings", flag.ContinueOnError)
-	return fs
-}
-
-type pluginMeta struct {
-	timeout int
-	adc     bool
-
-	credentialsWatch bool
-	credentialsFile  string
-	targetAudience   string
-}
-
 func main() {
-	pluginMeta := &pluginMeta{}
-	fs := pluginMeta.flagSet()
-	fs.IntVar(&pluginMeta.timeout, "timeout", 10, "Request timeout in seconds")
-	fs.BoolVar(&pluginMeta.adc, "adc", false, "Use Google Application Default Credentials instead of ServiceAccount JSON")
-	fs.StringVar(&pluginMeta.credentialsFile, "credentials-file", "", "Location of the JSON file with the application credentials")
-	fs.BoolVar(&pluginMeta.credentialsWatch, "credentials-watch", true, "Watch credential for reload")
-	fs.StringVar(&pluginMeta.targetAudience, "target-audience", "", "URI of audience claim")
-
-	fs.Parse(os.Args[1:])
-
-	options := googleidprovider.TokenProviderOptions{
-		Timeout:          pluginMeta.timeout,
-		Adc:              pluginMeta.adc,
-		CredentialsWatch: pluginMeta.credentialsWatch,
-		CredentialsFile:  pluginMeta.credentialsFile,
-		TargetAudience:   pluginMeta.targetAudience,
-	}
-
-	tokenProvider, err := googleidprovider.NewTokenProvider(options)
+	tokenProvider, err := new(googleidprovider.Factory).New(os.Args[1:])
 	if err != nil {
-		logrus.Errorf("cannot initialize googleid-token provider: %v", err)
+		logrus.Errorf("cannot initialize google-id-token provider: %v", err)
 		os.Exit(1)
 	}
 

pkg/apis/gateway.go

@@ -18,6 +18,10 @@ type TokenProvider interface {
 	GetToken(ctx context.Context, request TokenRequest) (TokenResponse, error)
 }
 
+type TokenProviderFactory interface {
+	New(params []string) (TokenProvider, error)
+}
+
 type VerifyRequest struct {
 	Token  string
 	Params []string
@@ -32,3 +36,7 @@ type TokenInfo interface {
 	// VerifyToken verifies the providing token and performs authorization. The returned error is only used by the underlying rpc protocol
 	VerifyToken(ctx context.Context, request VerifyRequest) (VerifyResponse, error)
 }
+
+type TokenInfoFactory interface {
+	New(params []string) (TokenInfo, error)
+}

pkg/apis/localauth.go

@@ -3,3 +3,7 @@ package apis
 type PasswordAuthenticator interface {
 	Authenticate(username, password string) (bool, int32, error)
 }
+
+type PasswordAuthenticatorFactory interface {
+	New(params []string) (PasswordAuthenticator, error)
+}