Adding middleware to inject Auth token for internal requests to front…

…end (cadence-workflow#4364)

client/clientBean.go

@@ -31,6 +31,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	clientworker "go.uber.org/cadence/worker"
 	"go.uber.org/yarpc"
 	"go.uber.org/yarpc/api/peer"
 	"go.uber.org/yarpc/api/transport"
@@ -42,6 +43,8 @@ import (
 	"github.com/uber/cadence/client/frontend"
 	"github.com/uber/cadence/client/history"
 	"github.com/uber/cadence/client/matching"
+	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/authorization"
 	"github.com/uber/cadence/common/cluster"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
@@ -66,10 +69,14 @@ type (
 		SetRemoteFrontendClient(cluster string, client frontend.Client)
 	}
 
+	DispatcherOptions struct {
+		AuthProvider clientworker.AuthorizationProvider
+	}
+
 	// DispatcherProvider provides a dispatcher to a given address
 	DispatcherProvider interface {
-		GetTChannel(name string, address string) (*yarpc.Dispatcher, error)
-		GetGRPC(name string, address string) (*yarpc.Dispatcher, error)
+		GetTChannel(name string, address string, options *DispatcherOptions) (*yarpc.Dispatcher, error)
+		GetGRPC(name string, address string, options *DispatcherOptions) (*yarpc.Dispatcher, error)
 	}
 
 	clientBeanImpl struct {
@@ -119,14 +126,26 @@ func NewClientBean(factory Factory, dispatcherProvider DispatcherProvider, clust
 			continue
 		}
 
+		var dispatcherOptions *DispatcherOptions
+		if info.AuthorizationProvider.Enable {
+			authProvider, err := authorization.GetAuthProviderClient(info.AuthorizationProvider.PrivateKey)
+			if err != nil {
+				return nil, err
+			}
+			dispatcherOptions = &DispatcherOptions{
+				AuthProvider: authProvider,
+			}
+		}
+
 		var dispatcher *yarpc.Dispatcher
 		var err error
 		switch info.RPCTransport {
 		case tchannel.TransportName:
-			dispatcher, err = dispatcherProvider.GetTChannel(info.RPCName, info.RPCAddress)
+			dispatcher, err = dispatcherProvider.GetTChannel(info.RPCName, info.RPCAddress, dispatcherOptions)
 		case grpc.TransportName:
-			dispatcher, err = dispatcherProvider.GetGRPC(info.RPCName, info.RPCAddress)
+			dispatcher, err = dispatcherProvider.GetGRPC(info.RPCName, info.RPCAddress, dispatcherOptions)
 		}
+
 		if err != nil {
 			return nil, err
 		}
@@ -265,7 +284,7 @@ func NewDNSYarpcDispatcherProvider(logger log.Logger, interval time.Duration) Di
 	}
 }
 
-func (p *dnsDispatcherProvider) GetTChannel(serviceName string, address string) (*yarpc.Dispatcher, error) {
+func (p *dnsDispatcherProvider) GetTChannel(serviceName string, address string, options *DispatcherOptions) (*yarpc.Dispatcher, error) {
 	tchanTransport, err := tchannel.NewTransport(
 		tchannel.ServiceName(serviceName),
 		// this aim to get rid of the annoying popup about accepting incoming network connections
@@ -284,10 +303,10 @@ func (p *dnsDispatcherProvider) GetTChannel(serviceName string, address string)
 	outbound := tchanTransport.NewOutbound(peerList)
 
 	p.logger.Info("Creating TChannel dispatcher outbound", tag.Address(address))
-	return p.createOutboundDispatcher(serviceName, outbound)
+	return p.createOutboundDispatcher(serviceName, outbound, options)
 }
 
-func (p *dnsDispatcherProvider) GetGRPC(serviceName string, address string) (*yarpc.Dispatcher, error) {
+func (p *dnsDispatcherProvider) GetGRPC(serviceName string, address string, options *DispatcherOptions) (*yarpc.Dispatcher, error) {
 	grpcTransport := grpc.NewTransport()
 
 	peerList := roundrobin.New(grpcTransport)
@@ -299,27 +318,50 @@ func (p *dnsDispatcherProvider) GetGRPC(serviceName string, address string) (*ya
 	outbound := grpcTransport.NewOutbound(peerList)
 
 	p.logger.Info("Creating GRPC dispatcher outbound", tag.Address(address))
-	return p.createOutboundDispatcher(serviceName, outbound)
+	return p.createOutboundDispatcher(serviceName, outbound, options)
 }
 
-func (p *dnsDispatcherProvider) createOutboundDispatcher(serviceName string, outbound transport.UnaryOutbound) (*yarpc.Dispatcher, error) {
-	// Attach the outbound to the dispatcher (this will add middleware/logging/etc)
-	dispatcher := yarpc.NewDispatcher(yarpc.Config{
+func (p *dnsDispatcherProvider) createOutboundDispatcher(serviceName string, outbound transport.UnaryOutbound, options *DispatcherOptions) (*yarpc.Dispatcher, error) {
+	cfg := yarpc.Config{
 		Name: crossDCCaller,
 		Outbounds: yarpc.Outbounds{
 			serviceName: transport.Outbounds{
 				Unary:       outbound,
 				ServiceName: serviceName,
 			},
 		},
-	})
+	}
+	if options != nil && options.AuthProvider != nil {
+		cfg.OutboundMiddleware = yarpc.OutboundMiddleware{
+			Unary: &outboundMiddleware{authProvider: options.AuthProvider},
+		}
+	}
+
+	// Attach the outbound to the dispatcher (this will add middleware/logging/etc)
+	dispatcher := yarpc.NewDispatcher(cfg)
 
 	if err := dispatcher.Start(); err != nil {
 		return nil, err
 	}
 	return dispatcher, nil
 }
 
+type outboundMiddleware struct {
+	authProvider clientworker.AuthorizationProvider
+}
+
+func (om *outboundMiddleware) Call(ctx context.Context, request *transport.Request, out transport.UnaryOutbound) (*transport.Response, error) {
+	if om.authProvider != nil {
+		token, err := om.authProvider.GetAuthToken()
+		if err != nil {
+			return nil, err
+		}
+		request.Headers = request.Headers.
+			With(common.AuthorizationTokenHeaderName, string(token))
+	}
+	return out.Call(ctx, request)
+}
+
 func newDNSUpdater(list peer.List, dnsPort string, interval time.Duration, logger log.Logger) (*dnsUpdater, error) {
 	ss := strings.Split(dnsPort, ":")
 	if len(ss) != 2 {

cmd/server/cadence/server.go

@@ -30,6 +30,7 @@ import (
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/archiver"
 	"github.com/uber/cadence/common/archiver/provider"
+	"github.com/uber/cadence/common/authorization"
 	"github.com/uber/cadence/common/blobstore/filestore"
 	"github.com/uber/cadence/common/cluster"
 	"github.com/uber/cadence/common/config"
@@ -213,7 +214,18 @@ func (s *server) startService() common.Daemon {
 		}
 	}
 
-	dispatcher, err := params.DispatcherProvider.GetTChannel(common.FrontendServiceName, s.cfg.PublicClient.HostPort)
+	var options *client.DispatcherOptions
+	if s.cfg.Authorization.OAuthAuthorizer.Enable {
+		clusterName := s.cfg.ClusterGroupMetadata.CurrentClusterName
+		authProvider, err := authorization.GetAuthProviderClient(s.cfg.ClusterGroupMetadata.ClusterGroup[clusterName].AuthorizationProvider.PrivateKey)
+		if err != nil {
+			log.Fatalf("failed to create AuthProvider: %v", err.Error())
+		}
+		options = &client.DispatcherOptions{
+			AuthProvider: authProvider,
+		}
+	}
+	dispatcher, err := params.DispatcherProvider.GetTChannel(common.FrontendServiceName, s.cfg.PublicClient.HostPort, options)
 	if err != nil {
 		log.Fatalf("failed to construct dispatcher: %v", err)
 	}

common/authorization/authorizer.go

@@ -24,6 +24,10 @@ package authorization
 
 import (
 	"context"
+	"fmt"
+	"io/ioutil"
+
+	clientworker "go.uber.org/cadence/worker"
 
 	"github.com/uber/cadence/common/types"
 )
@@ -84,3 +88,11 @@ func NewPermission(permission string) Permission {
 type Authorizer interface {
 	Authorize(ctx context.Context, attributes *Attributes) (Result, error)
 }
+
+func GetAuthProviderClient(privateKey string) (clientworker.AuthorizationProvider, error) {
+	pk, err := ioutil.ReadFile(privateKey)
+	if err != nil {
+		return nil, fmt.Errorf("invalid private key path %s", privateKey)
+	}
+	return clientworker.NewAdminJwtAuthorizationProvider(pk), nil
+}

common/authorization/factory_test.go

@@ -62,9 +62,8 @@ func cfgOAuth() config.Authorization {
 		OAuthAuthorizer: config.OAuthAuthorizer{
 			Enable: true,
 			JwtCredentials: config.JwtCredentials{
-				Algorithm:  jwt.RS256.String(),
-				PublicKey:  "public",
-				PrivateKey: "private",
+				Algorithm: jwt.RS256.String(),
+				PublicKey: "public",
 			},
 			MaxJwtTTL: 12345,
 		},

common/authorization/oauthAuthorizer.go

@@ -78,6 +78,10 @@ func (a *oauthAuthority) Authorize(
 		return Result{Decision: DecisionDeny}, err
 	}
 	token := call.Header(common.AuthorizationTokenHeaderName)
+	if token == "" {
+		a.log.Debug("request is not authorized", tag.Error(fmt.Errorf("token is not set in header")))
+		return Result{Decision: DecisionDeny}, nil
+	}
 	claims, err := a.parseToken(token, verifier)
 	if err != nil {
 		a.log.Debug("request is not authorized", tag.Error(err))

common/authorization/oauthAutorizer_test.go

@@ -64,9 +64,8 @@ func (s *oauthSuite) SetupTest() {
 	s.cfg = config.OAuthAuthorizer{
 		Enable: true,
 		JwtCredentials: config.JwtCredentials{
-			Algorithm:  jwt.RS256.String(),
-			PublicKey:  "../../config/credentials/keytest.pub",
-			PrivateKey: "../../config/credentials/keytest",
+			Algorithm: jwt.RS256.String(),
+			PublicKey: "../../config/credentials/keytest.pub",
 		},
 		MaxJwtTTL: 300000001,
 	}
@@ -139,6 +138,21 @@ func (s *oauthSuite) TestItIsAdmin() {
 	s.Equal(result.Decision, DecisionAllow)
 }
 
+func (s *oauthSuite) TestEmptyToken() {
+	ctx := context.Background()
+	ctx, call := encoding.NewInboundCall(ctx)
+	err := call.ReadFromRequest(&transport.Request{
+		Headers: transport.NewHeaders().With(common.AuthorizationTokenHeaderName, ""),
+	})
+	s.NoError(err)
+	authorizer := NewOAuthAuthorizer(s.cfg, s.logger, s.domainCache)
+	s.logger.On("Debug", "request is not authorized", mock.MatchedBy(func(t []tag.Tag) bool {
+		return fmt.Sprintf("%v", t[0].Field().Interface) == "token is not set in header"
+	}))
+	result, _ := authorizer.Authorize(ctx, &s.att)
+	s.Equal(result.Decision, DecisionDeny)
+}
+
 func (s *oauthSuite) TestGetDomainError() {
 	s.domainCache.EXPECT().GetDomain(s.att.DomainName).Return(nil, fmt.Errorf("error")).Times(1)
 	authorizer := NewOAuthAuthorizer(s.cfg, s.logger, s.domainCache)

common/config/authorization.go

@@ -47,9 +47,6 @@ func (a *Authorization) validateOAuth() error {
 	if oauthConfig.MaxJwtTTL <= 0 {
 		return fmt.Errorf("[OAuthConfig] MaxTTL must be greater than 0")
 	}
-	if oauthConfig.JwtCredentials.PrivateKey == "" {
-		return fmt.Errorf("[OAuthConfig] PrivateKey can't be empty")
-	}
 	if oauthConfig.JwtCredentials.PublicKey == "" {
 		return fmt.Errorf("[OAuthConfig] PublicKey can't be empty")
 	}

common/config/authorization_test.go

@@ -56,34 +56,13 @@ func TestTTLIsZero(t *testing.T) {
 	assert.EqualError(t, err, "[OAuthConfig] MaxTTL must be greater than 0")
 }
 
-func TestPrivateKeyIsEmpty(t *testing.T) {
-	cfg := Authorization{
-		OAuthAuthorizer: OAuthAuthorizer{
-			Enable: true,
-			JwtCredentials: JwtCredentials{
-				Algorithm:  "",
-				PublicKey:  "",
-				PrivateKey: "",
-			},
-			MaxJwtTTL: 1000000,
-		},
-		NoopAuthorizer: NoopAuthorizer{
-			Enable: false,
-		},
-	}
-
-	err := cfg.Validate()
-	assert.EqualError(t, err, "[OAuthConfig] PrivateKey can't be empty")
-}
-
 func TestPublicKeyIsEmpty(t *testing.T) {
 	cfg := Authorization{
 		OAuthAuthorizer: OAuthAuthorizer{
 			Enable: true,
 			JwtCredentials: JwtCredentials{
-				Algorithm:  "",
-				PublicKey:  "",
-				PrivateKey: "private",
+				Algorithm: "",
+				PublicKey: "",
 			},
 			MaxJwtTTL: 1000000,
 		},
@@ -101,9 +80,8 @@ func TestAlgorithmIsInvalid(t *testing.T) {
 		OAuthAuthorizer: OAuthAuthorizer{
 			Enable: true,
 			JwtCredentials: JwtCredentials{
-				Algorithm:  "SHA256",
-				PublicKey:  "public",
-				PrivateKey: "private",
+				Algorithm: "SHA256",
+				PublicKey: "public",
 			},
 			MaxJwtTTL: 1000000,
 		},
@@ -121,9 +99,8 @@ func TestCorrectValidation(t *testing.T) {
 		OAuthAuthorizer: OAuthAuthorizer{
 			Enable: true,
 			JwtCredentials: JwtCredentials{
-				Algorithm:  "RS256",
-				PublicKey:  "public",
-				PrivateKey: "private",
+				Algorithm: "RS256",
+				PublicKey: "public",
 			},
 			MaxJwtTTL: 1000000,
 		},

common/config/cluster.go

@@ -66,6 +66,17 @@ type (
 		// Allowed values: tchannel|grpc
 		// Default: tchannel
 		RPCTransport string `yaml:"rpcTransport"`
+		// AuthorizationProvider contains the information to authorize the cluster
+		AuthorizationProvider AuthorizationProvider `yaml:"authorizationProvider"`
+	}
+
+	AuthorizationProvider struct {
+		// Enable indicates if the auth provider is enabled
+		Enable bool `yaml:"enable"`
+		// Type auth provider type
+		Type string `yaml:"type"` // only supports OAuthAuthorization
+		// PrivateKey is the private key path
+		PrivateKey string `yaml:"privateKey"`
 	}
 )
 

common/config/config.go

@@ -101,8 +101,6 @@ type (
 		Algorithm string `yaml:"algorithm"`
 		// Public Key Path for verifying JWT token passed in from external clients
 		PublicKey string `yaml:"publicKey"`
-		// Private Key Path for creating JWT token
-		PrivateKey string `yaml:"privateKey"`
 	}
 
 	// Service contains the service specific config items