Implementation of JWTAuthorizationProvider

go.mod

@@ -4,6 +4,7 @@ go 1.13
 
 require (
 	github.com/apache/thrift v0.0.0-20161221203622-b2a4d4ae21c7
+	github.com/cristalhq/jwt/v3 v3.1.0 // indirect
 	github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/mock v1.4.4

go.sum

@@ -24,6 +24,8 @@ github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGX
 github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd h1:qMd81Ts1T2OTKmB4acZcyKaMtRnY5Y44NuXGX2GFJ1w=
 github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd/go.mod h1:sE/e/2PUdi/liOCUjSTXgM1o87ZssimdTWN964YiIeI=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/cristalhq/jwt/v3 v3.1.0 h1:iLeL9VzB0SCtjCy9Kg53rMwTcrNm+GHyVcz2eUujz6s=
+github.com/cristalhq/jwt/v3 v3.1.0/go.mod h1:XOnIXst8ozq/esy5N1XOlSyQqBd+84fxJ99FK+1jgL8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

internal/client.go

@@ -24,6 +24,8 @@ package internal
 import (
 	"context"
 	"fmt"
+	"go.uber.org/cadence/internal/common/auth"
+	"go.uber.org/cadence/worker"
 	"time"
 
 	"github.com/opentracing/opentracing-go"
@@ -336,6 +338,7 @@ type (
 		Tracer             opentracing.Tracer
 		ContextPropagators []ContextPropagator
 		FeatureFlags       FeatureFlags
+		Authorization 	   worker.AuthorizationProvider
 	}
 
 	// StartWorkflowOptions configuration parameters for starting a workflow execution.
@@ -542,9 +545,12 @@ func NewClient(service workflowserviceclient.Interface, domain string, options *
 	} else {
 		tracer = opentracing.NoopTracer{}
 	}
-
+	if options.Authorization != nil{
+		service = auth.NewWorkflowServiceWrapper(service, options.Authorization)
+	}
+	service = metrics.NewWorkflowServiceWrapper(service, metricScope)
 	return &workflowClient{
-		workflowService:    metrics.NewWorkflowServiceWrapper(service, metricScope),
+		workflowService:    service,
 		domain:             domain,
 		registry:           newRegistry(),
 		metricsScope:       metrics.NewTaggedScope(metricScope),
@@ -569,9 +575,12 @@ func NewDomainClient(service workflowserviceclient.Interface, options *ClientOpt
 		metricScope = options.MetricsScope
 	}
 	metricScope = tagScope(metricScope, tagDomain, "domain-client", clientImplHeaderName, clientImplHeaderValue)
-
+	if options.Authorization != nil{
+		service = auth.NewWorkflowServiceWrapper(service, options.Authorization)
+	}
+	service = metrics.NewWorkflowServiceWrapper(service, metricScope)
 	return &domainClient{
-		workflowService: metrics.NewWorkflowServiceWrapper(service, metricScope),
+		workflowService: service,
 		metricsScope:    metricScope,
 		identity:        identity,
 		featureFlags:    getFeatureFlags(options),

internal/common/auth/service_wrapper.go

@@ -0,0 +1,280 @@
+// Copyright (c) 2021 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package auth
+
+import (
+    "context"
+    "go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
+    "go.uber.org/cadence/.gen/go/shared"
+    "go.uber.org/cadence/worker"
+    "go.uber.org/yarpc"
+)
+
+const (
+    jwtHeaderName = "cadence-authorization"
+)
+
+type workflowServiceAuthWrapper struct {
+    service workflowserviceclient.Interface
+    authProvider worker.AuthorizationProvider
+}
+
+// NewWorkflowServiceWrapper creates
+func NewWorkflowServiceWrapper(service workflowserviceclient.Interface, authorizationProvider worker.AuthorizationProvider) workflowserviceclient.Interface {
+    return &workflowServiceAuthWrapper{
+        service: service,
+        authProvider: authorizationProvider,
+    }
+}
+
+func (w *workflowServiceAuthWrapper) getYarpcJWTHeader() yarpc.CallOption {
+    token := w.authProvider.GetAuthToken()
+    return yarpc.WithHeader(jwtHeaderName, string(token))
+}
+
+func (w *workflowServiceAuthWrapper) DeprecateDomain(ctx context.Context, request *shared.DeprecateDomainRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.DeprecateDomain(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) ListDomains(ctx context.Context, request *shared.ListDomainsRequest, opts ...yarpc.CallOption) (*shared.ListDomainsResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ListDomains(ctx, request, opts...)
+
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) DescribeDomain(ctx context.Context, request *shared.DescribeDomainRequest, opts ...yarpc.CallOption) (*shared.DescribeDomainResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.DescribeDomain(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) DescribeWorkflowExecution(ctx context.Context, request *shared.DescribeWorkflowExecutionRequest, opts ...yarpc.CallOption) (*shared.DescribeWorkflowExecutionResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.DescribeWorkflowExecution(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) GetWorkflowExecutionHistory(ctx context.Context, request *shared.GetWorkflowExecutionHistoryRequest, opts ...yarpc.CallOption) (*shared.GetWorkflowExecutionHistoryResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.GetWorkflowExecutionHistory(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) ListClosedWorkflowExecutions(ctx context.Context, request *shared.ListClosedWorkflowExecutionsRequest, opts ...yarpc.CallOption) (*shared.ListClosedWorkflowExecutionsResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ListClosedWorkflowExecutions(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) ListOpenWorkflowExecutions(ctx context.Context, request *shared.ListOpenWorkflowExecutionsRequest, opts ...yarpc.CallOption) (*shared.ListOpenWorkflowExecutionsResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ListOpenWorkflowExecutions(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) ListWorkflowExecutions(ctx context.Context, request *shared.ListWorkflowExecutionsRequest, opts ...yarpc.CallOption) (*shared.ListWorkflowExecutionsResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ListWorkflowExecutions(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) ListArchivedWorkflowExecutions(ctx context.Context, request *shared.ListArchivedWorkflowExecutionsRequest, opts ...yarpc.CallOption) (*shared.ListArchivedWorkflowExecutionsResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ListArchivedWorkflowExecutions(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) ScanWorkflowExecutions(ctx context.Context, request *shared.ListWorkflowExecutionsRequest, opts ...yarpc.CallOption) (*shared.ListWorkflowExecutionsResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ScanWorkflowExecutions(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) CountWorkflowExecutions(ctx context.Context, request *shared.CountWorkflowExecutionsRequest, opts ...yarpc.CallOption) (*shared.CountWorkflowExecutionsResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.CountWorkflowExecutions(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) PollForActivityTask(ctx context.Context, request *shared.PollForActivityTaskRequest, opts ...yarpc.CallOption) (*shared.PollForActivityTaskResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.PollForActivityTask(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) PollForDecisionTask(ctx context.Context, request *shared.PollForDecisionTaskRequest, opts ...yarpc.CallOption) (*shared.PollForDecisionTaskResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.PollForDecisionTask(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) RecordActivityTaskHeartbeat(ctx context.Context, request *shared.RecordActivityTaskHeartbeatRequest, opts ...yarpc.CallOption) (*shared.RecordActivityTaskHeartbeatResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.RecordActivityTaskHeartbeat(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) RecordActivityTaskHeartbeatByID(ctx context.Context, request *shared.RecordActivityTaskHeartbeatByIDRequest, opts ...yarpc.CallOption) (*shared.RecordActivityTaskHeartbeatResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.RecordActivityTaskHeartbeatByID(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) RegisterDomain(ctx context.Context, request *shared.RegisterDomainRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RegisterDomain(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) RequestCancelWorkflowExecution(ctx context.Context, request *shared.RequestCancelWorkflowExecutionRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RequestCancelWorkflowExecution(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) RespondActivityTaskCanceled(ctx context.Context, request *shared.RespondActivityTaskCanceledRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RespondActivityTaskCanceled(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) RespondActivityTaskCompleted(ctx context.Context, request *shared.RespondActivityTaskCompletedRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RespondActivityTaskCompleted(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) RespondActivityTaskFailed(ctx context.Context, request *shared.RespondActivityTaskFailedRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RespondActivityTaskFailed(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) RespondActivityTaskCanceledByID(ctx context.Context, request *shared.RespondActivityTaskCanceledByIDRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RespondActivityTaskCanceledByID(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) RespondActivityTaskCompletedByID(ctx context.Context, request *shared.RespondActivityTaskCompletedByIDRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RespondActivityTaskCompletedByID(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) RespondActivityTaskFailedByID(ctx context.Context, request *shared.RespondActivityTaskFailedByIDRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RespondActivityTaskFailedByID(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) RespondDecisionTaskCompleted(ctx context.Context, request *shared.RespondDecisionTaskCompletedRequest, opts ...yarpc.CallOption) (*shared.RespondDecisionTaskCompletedResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    response, err := w.service.RespondDecisionTaskCompleted(ctx, request, opts...)
+    return response, err
+}
+
+func (w *workflowServiceAuthWrapper) RespondDecisionTaskFailed(ctx context.Context, request *shared.RespondDecisionTaskFailedRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RespondDecisionTaskFailed(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) SignalWorkflowExecution(ctx context.Context, request *shared.SignalWorkflowExecutionRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.SignalWorkflowExecution(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) SignalWithStartWorkflowExecution(ctx context.Context, request *shared.SignalWithStartWorkflowExecutionRequest, opts ...yarpc.CallOption) (*shared.StartWorkflowExecutionResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.SignalWithStartWorkflowExecution(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) StartWorkflowExecution(ctx context.Context, request *shared.StartWorkflowExecutionRequest, opts ...yarpc.CallOption) (*shared.StartWorkflowExecutionResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.StartWorkflowExecution(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) TerminateWorkflowExecution(ctx context.Context, request *shared.TerminateWorkflowExecutionRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.TerminateWorkflowExecution(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) ResetWorkflowExecution(ctx context.Context, request *shared.ResetWorkflowExecutionRequest, opts ...yarpc.CallOption) (*shared.ResetWorkflowExecutionResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ResetWorkflowExecution(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) UpdateDomain(ctx context.Context, request *shared.UpdateDomainRequest, opts ...yarpc.CallOption) (*shared.UpdateDomainResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.UpdateDomain(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) QueryWorkflow(ctx context.Context, request *shared.QueryWorkflowRequest, opts ...yarpc.CallOption) (*shared.QueryWorkflowResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.QueryWorkflow(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) ResetStickyTaskList(ctx context.Context, request *shared.ResetStickyTaskListRequest, opts ...yarpc.CallOption) (*shared.ResetStickyTaskListResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ResetStickyTaskList(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) DescribeTaskList(ctx context.Context, request *shared.DescribeTaskListRequest, opts ...yarpc.CallOption) (*shared.DescribeTaskListResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.DescribeTaskList(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) RespondQueryTaskCompleted(ctx context.Context, request *shared.RespondQueryTaskCompletedRequest, opts ...yarpc.CallOption) error {
+    opts = append(opts, w.getYarpcJWTHeader())
+    err := w.service.RespondQueryTaskCompleted(ctx, request, opts...)
+    return err
+}
+
+func (w *workflowServiceAuthWrapper) GetSearchAttributes(ctx context.Context, opts ...yarpc.CallOption) (*shared.GetSearchAttributesResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.GetSearchAttributes(ctx, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) ListTaskListPartitions(ctx context.Context, request *shared.ListTaskListPartitionsRequest, opts ...yarpc.CallOption) (*shared.ListTaskListPartitionsResponse, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.ListTaskListPartitions(ctx, request, opts...)
+    return result, err
+}
+
+func (w *workflowServiceAuthWrapper) GetClusterInfo(ctx context.Context, opts ...yarpc.CallOption) (*shared.ClusterInfo, error) {
+    opts = append(opts, w.getYarpcJWTHeader())
+    result, err := w.service.GetClusterInfo(ctx, opts...)
+    return result, err
+}