Implement unsecured-jwt-info plugin

Author: everesio

Makefile

@@ -64,7 +64,10 @@ plugin.google-id-provider:
 plugin.google-id-info:
 	CGO_ENABLED=0 go build -o build/google-id-info $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-googleid-info/main.go
 
-all: build plugin.auth-user plugin.auth-ldap plugin.google-id-provider plugin.google-id-info
+plugin.unsecured-jwt-info:
+	CGO_ENABLED=0 go build -o build/unsecured-jwt-info $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" cmd/plugin-unsecured-jwt-info/main.go
+
+all: build plugin.auth-user plugin.auth-ldap plugin.google-id-provider plugin.google-id-info plugin.unsecured-jwt-info
 
 clean:
 	@rm -rf build

README.md

@@ -170,6 +170,14 @@ See:
                              --auth-local-param "--user-attr=uid" \
                              --bootstrap-server-mapping "192.168.99.100:32400,127.0.0.1:32400"
 
+    make clean build plugin.unsecured-jwt-info && build/kafka-proxy server \
+                             --auth-local-enable \
+                             --auth-local-command build/unsecured-jwt-info \
+                             --auth-local-mechanism "OAUTHBEARER" \
+                             --auth-local-param "--claim-sub=alice" \
+                             --auth-local-param "--claim-sub=bob" \
+                             --bootstrap-server-mapping "192.168.99.100:32400,127.0.0.1:32400"
+
 ### Kafka Gateway example
 
 Authentication between Kafka Proxy Client and Kafka Proxy Server with Google-ID (service account JWT)

cmd/plugin-unsecured-jwt-info/main.go

@@ -0,0 +1,154 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"flag"
+	"github.com/grepplabs/kafka-proxy/pkg/apis"
+	"github.com/grepplabs/kafka-proxy/pkg/libs/util"
+	"github.com/grepplabs/kafka-proxy/plugin/token-info/shared"
+	"github.com/hashicorp/go-plugin"
+	"github.com/sirupsen/logrus"
+	"os"
+	"strings"
+	"time"
+)
+
+const (
+	StatusOK                      = 0
+	StatusEmptyToken              = 1
+	StatusParseJWTFailed          = 2
+	StatusWrongAlgorithm          = 3
+	StatusUnauthorized            = 4
+	StatusNoIssueTimeInToken      = 5
+	StatusNoExpirationTimeInToken = 6
+	StatusTokenTooEarly           = 7
+	StatusTokenExpired            = 8
+
+	AlgorithmNone = "none"
+)
+
+var (
+	clockSkew = 1 * time.Minute
+)
+
+type UnsecuredJWTVerifier struct {
+	claimSub map[string]struct{}
+}
+
+type pluginMeta struct {
+	claimSub util.ArrayFlags
+}
+
+func (f *pluginMeta) flagSet() *flag.FlagSet {
+	fs := flag.NewFlagSet("unsecured-jwt-info info settings", flag.ContinueOnError)
+	return fs
+}
+
+// Implements apis.TokenInfo
+func (v UnsecuredJWTVerifier) VerifyToken(ctx context.Context, request apis.VerifyRequest) (apis.VerifyResponse, error) {
+	if request.Token == "" {
+		return getVerifyResponseResponse(StatusEmptyToken)
+	}
+
+	header, claimSet, err := Decode(request.Token)
+	if err != nil {
+		return getVerifyResponseResponse(StatusParseJWTFailed)
+	}
+	if header.Algorithm != AlgorithmNone {
+		return getVerifyResponseResponse(StatusWrongAlgorithm)
+	}
+
+	if len(v.claimSub) != 0 {
+		if _, ok := v.claimSub[claimSet.Sub]; !ok {
+			return getVerifyResponseResponse(StatusUnauthorized)
+		}
+	}
+	if claimSet.Iat < 1 {
+		return getVerifyResponseResponse(StatusNoIssueTimeInToken)
+	}
+	if claimSet.Exp < 1 {
+		return getVerifyResponseResponse(StatusNoExpirationTimeInToken)
+	}
+
+	earliest := int64(claimSet.Iat) - int64(clockSkew.Seconds())
+	latest := int64(claimSet.Exp) + int64(clockSkew.Seconds())
+	unix := time.Now().Unix()
+
+	if unix < earliest {
+		return getVerifyResponseResponse(StatusTokenTooEarly)
+	}
+	if unix > latest {
+		return getVerifyResponseResponse(StatusTokenExpired)
+	}
+	return getVerifyResponseResponse(StatusOK)
+}
+
+type Header struct {
+	Algorithm string `json:"alg"`
+}
+
+// kafka client sends float instead of int
+type ClaimSet struct {
+	Sub         string                 `json:"sub,omitempty"`
+	Exp         float64                `json:"exp"`
+	Iat         float64                `json:"iat"`
+	OtherClaims map[string]interface{} `json:"-"`
+}
+
+func Decode(token string) (*Header, *ClaimSet, error) {
+	args := strings.Split(token, ".")
+	if len(args) < 2 {
+		return nil, nil, errors.New("jws: invalid token received")
+	}
+	decodedHeader, err := base64.RawURLEncoding.DecodeString(args[0])
+	if err != nil {
+		return nil, nil, err
+	}
+	decodedPayload, err := base64.RawURLEncoding.DecodeString(args[1])
+	if err != nil {
+		return nil, nil, err
+	}
+
+	header := &Header{}
+	err = json.NewDecoder(bytes.NewBuffer(decodedHeader)).Decode(header)
+	if err != nil {
+		return nil, nil, err
+	}
+	claimSet := &ClaimSet{}
+	err = json.NewDecoder(bytes.NewBuffer(decodedPayload)).Decode(claimSet)
+	if err != nil {
+		return nil, nil, err
+	}
+	return header, claimSet, nil
+}
+
+func getVerifyResponseResponse(status int) (apis.VerifyResponse, error) {
+	success := status == StatusOK
+	return apis.VerifyResponse{Success: success, Status: int32(status)}, nil
+}
+
+func main() {
+	pluginMeta := &pluginMeta{}
+	fs := pluginMeta.flagSet()
+	fs.Var(&pluginMeta.claimSub, "claim-sub", "Allowed subject claim (user name)")
+	fs.Parse(os.Args[1:])
+
+	logrus.Infof("Unsecured JWT sub claims: %v", pluginMeta.claimSub)
+
+	unsecuredJWTVerifier := &UnsecuredJWTVerifier{
+		claimSub: pluginMeta.claimSub.AsMap(),
+	}
+
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: shared.Handshake,
+		Plugins: map[string]plugin.Plugin{
+			"unsecuredJWTInfo": &shared.TokenInfoPlugin{Impl: unsecuredJWTVerifier},
+		},
+		// A non-nil value here enables gRPC serving for this plugin...
+		GRPCServer: plugin.DefaultGRPCServer,
+	})
+}

pkg/libs/googleid-info/factory.go

@@ -2,8 +2,8 @@ package googleidinfo
 
 import (
 	"flag"
-	"fmt"
 	"github.com/grepplabs/kafka-proxy/pkg/apis"
+	"github.com/grepplabs/kafka-proxy/pkg/libs/util"
 	"github.com/grepplabs/kafka-proxy/pkg/registry"
 )
 
@@ -20,27 +20,8 @@ func (f *pluginMeta) flagSet() *flag.FlagSet {
 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
+	audience             util.ArrayFlags
+	emailsRegex          util.ArrayFlags
 }
 
 type Factory struct {

pkg/libs/util/flags.go

@@ -0,0 +1,22 @@
+package util
+
+import "fmt"
+
+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
+}

proxy/auth.go

@@ -118,7 +118,7 @@ func (b *AuthServer) receiveAndSendGatewayAuth(conn DeadlineReaderWriter) error
 		return err
 	}
 	if !resp.Success {
-		return fmt.Errorf("verify token failed with status: %d", resp.Status)
+		return fmt.Errorf("gateway server verify token failed with status: %d", resp.Status)
 	}
 
 	logrus.Debugf("gateway handshake payload: %s", data)

proxy/sasl_local_auth.go

@@ -70,7 +70,7 @@ func (p *LocalSaslOauth) doLocalAuth(saslAuthBytes []byte) (err error) {
 		return err
 	}
 	if !resp.Success {
-		return fmt.Errorf("verify token failed with status: %d", resp.Status)
+		return fmt.Errorf("local oauth verify token failed with status: %d", resp.Status)
 	}
 	return nil
 }