gateway-client/server stubs

Author: everesio

Makefile

@@ -79,11 +79,20 @@ release: clean build.linux build/osx/$(BINARY)
 protoc.local-auth:
 	protoc -I plugin/local-auth/proto/ plugin/local-auth/proto/auth.proto --go_out=plugins=grpc:plugin/local-auth/proto/
 
+protoc.gateway-client:
+	protoc -I plugin/gateway-client/proto/ plugin/gateway-client/proto/token-provider.proto --go_out=plugins=grpc:plugin/gateway-client/proto/
+
+protoc.gateway-server:
+	protoc -I plugin/gateway-server/proto/ plugin/gateway-server/proto/token-info.proto --go_out=plugins=grpc:plugin/gateway-server/proto/
+
 plugin.auth-user:
 	CGO_ENABLED=0 go build -o build/auth-user $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-auth-user/main.go
 
 plugin.auth-ldap:
 	CGO_ENABLED=0 go build -o build/auth-ldap $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-auth-ldap/main.go
 
+plugin.googleid-provider:
+	CGO_ENABLED=0 go build -o build/googleid-provider $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-googleid-provider/main.go
+
 clean:
 	@rm -rf build

cmd/kafka-proxy/server.go

@@ -19,7 +19,8 @@ import (
 	"time"
 
 	"errors"
-	"github.com/grepplabs/kafka-proxy/plugin/local-auth/shared"
+	gatewayclient "github.com/grepplabs/kafka-proxy/plugin/gateway-client/shared"
+	localauth "github.com/grepplabs/kafka-proxy/plugin/local-auth/shared"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-plugin"
 )
@@ -143,7 +144,7 @@ func init() {
 func Run(_ *cobra.Command, _ []string) {
 	logrus.Infof("Starting kafka-proxy version %s", config.Version)
 
-	var passwordAuthenticator shared.PasswordAuthenticator
+	var passwordAuthenticator localauth.PasswordAuthenticator
 	if c.Auth.Local.Enable {
 		client := NewLocalAuthPluginClient()
 		defer client.Kill()
@@ -157,11 +158,32 @@ func Run(_ *cobra.Command, _ []string) {
 			logrus.Fatal(err)
 		}
 		var ok bool
-		passwordAuthenticator, ok = raw.(shared.PasswordAuthenticator)
+		passwordAuthenticator, ok = raw.(localauth.PasswordAuthenticator)
 		if !ok {
 			logrus.Fatal(errors.New("unsupported plugin type"))
 		}
 	}
+
+	var tokenProvider gatewayclient.TokenProvider
+	if c.Auth.Gateway.Client.Enable {
+		client := NewGatewayClientPluginClient()
+		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.(gatewayclient.TokenProvider)
+		if !ok {
+			logrus.Fatal(errors.New("unsupported plugin type"))
+		}
+	}
+
 	var g group.Group
 	{
 		// All active connections are stored in this variable.
@@ -175,7 +197,7 @@ func Run(_ *cobra.Command, _ []string) {
 		if err != nil {
 			logrus.Fatal(err)
 		}
-		proxyClient, err := proxy.NewClient(connset, c, listeners.GetNetAddressMapping, passwordAuthenticator)
+		proxyClient, err := proxy.NewClient(connset, c, listeners.GetNetAddressMapping, passwordAuthenticator, tokenProvider)
 		if err != nil {
 			logrus.Fatal(err)
 		}
@@ -274,24 +296,32 @@ 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 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" {
 		jsonFormat = true
 	}
 	logger := hclog.New(&hclog.LoggerOptions{
 		Output:     os.Stdout,
-		Level:      hclog.LevelFromString(c.Auth.Local.LogLevel),
+		Level:      hclog.LevelFromString(logLevel),
 		Name:       "plugin",
 		JSONFormat: jsonFormat,
 		TimeFormat: time.RFC3339,
 	})
 
 	return plugin.NewClient(&plugin.ClientConfig{
-		HandshakeConfig: shared.Handshake,
-		Plugins:         shared.PluginMap,
+		HandshakeConfig: handshakeConfig,
+		Plugins:         plugins,
 		Logger:          logger,
-		Cmd:             exec.Command(c.Auth.Local.Command, c.Auth.Local.Parameters...),
+		Cmd:             exec.Command(command, params...),
 		AllowedProtocols: []plugin.Protocol{
 			plugin.ProtocolNetRPC, plugin.ProtocolGRPC},
 	})

cmd/plugin-googleid-provider/main.go

@@ -0,0 +1,65 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"github.com/grepplabs/kafka-proxy/plugin/gateway-client/shared"
+	"github.com/hashicorp/go-plugin"
+	"golang.org/x/oauth2/google"
+	"google.golang.org/api/oauth2/v2"
+	"os"
+	"time"
+)
+
+type TokenProvider struct {
+	timeout int
+}
+
+//TODO: caching, expiry
+//TODO: refresh in the half of time
+//TODO: send claims
+func (p TokenProvider) GetToken(claims []string) (int32, string, error) {
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(p.timeout)*time.Second)
+	defer cancel()
+
+	tokenSource, err := google.DefaultTokenSource(ctx, oauth2.UserinfoEmailScope)
+	if err != nil {
+		return tokenResponse(1, "")
+	}
+	token, err := tokenSource.Token()
+	if err != nil {
+		return tokenResponse(2, "")
+	}
+	if token.Extra("id_token") == nil {
+		return tokenResponse(3, "")
+	}
+	idToken := token.Extra("id_token").(string)
+	return tokenResponse(0, idToken)
+
+}
+
+func tokenResponse(status int32, token string) (int32, string, error) {
+	return status, token, nil
+}
+
+func (f *TokenProvider) flagSet() *flag.FlagSet {
+	fs := flag.NewFlagSet("google-id provider settings", flag.ContinueOnError)
+	return fs
+}
+
+func main() {
+	tokenProvider := &TokenProvider{}
+	fs := tokenProvider.flagSet()
+	fs.Parse(os.Args[1:])
+	fs.IntVar(&tokenProvider.timeout, "timeout", 5, "Request timeout")
+
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: shared.Handshake,
+		Plugins: map[string]plugin.Plugin{
+			"tokenProvider": &shared.TokenProviderPlugin{Impl: tokenProvider},
+		},
+		// A non-nil value here enables gRPC serving for this plugin...
+		GRPCServer: plugin.DefaultGRPCServer,
+	})
+}

pkg/apis/gateway.go

@@ -0,0 +1,5 @@
+package apis
+
+type TokenProvider interface {
+	GetToken(claims []string) (int32, string, error)
+}

pkg/apis/localauth.go

@@ -0,0 +1,5 @@
+package apis
+
+type PasswordAuthenticator interface {
+	Authenticate(username, password string) (bool, int32, error)
+}

plugin/gateway-client/proto/token-provider.pb.go

@@ -0,0 +1,16 @@
+syntax = "proto3";
+package proto;
+
+message TokenRequest {
+    repeated string claims = 1;
+}
+
+message TokenResponse {
+    int32 status = 1;
+    string token = 2;
+    // TODO: expiry ?
+}
+
+service TokenProvider {
+    rpc GetToken(TokenRequest) returns (TokenResponse);
+}