Add context.Context to callback.Validator interface

Thread context through to callback validation so that deployments can check caller identity when validating callbacks. The base validator does not use the context.

Author: fretz12

chasm/lib/activity/frontend.go

@@ -38,7 +38,7 @@ var ErrStandaloneActivityDisabled = serviceerror.NewUnimplemented("Standalone ac
 
 type frontendHandler struct {
 	FrontendHandler
-	callbackValidator *callback.Validator
+	callbackValidator callback.Validator
 	client            activitypb.ActivityServiceClient
 	config            *Config
 	logger            log.Logger
@@ -50,7 +50,7 @@ type frontendHandler struct {
 
 // NewFrontendHandler creates a new FrontendHandler instance for processing activity frontend requests.
 func NewFrontendHandler(
-	callbackValidator *callback.Validator,
+	callbackValidator callback.Validator,
 	client activitypb.ActivityServiceClient,
 	config *Config,
 	logger log.Logger,
@@ -95,7 +95,7 @@ func (h *frontendHandler) StartActivityExecution(ctx context.Context, req *workf
 		return nil, err
 	}
 
-	modifiedReq, err := h.validateAndPopulateStartRequest(req, namespaceID)
+	modifiedReq, err := h.validateAndPopulateStartRequest(ctx, req, namespaceID)
 	if err != nil {
 		return nil, err
 	}
@@ -351,6 +351,7 @@ func (h *frontendHandler) RequestCancelActivityExecution(
 }
 
 func (h *frontendHandler) validateAndPopulateStartRequest(
+	ctx context.Context,
 	req *workflowservice.StartActivityExecutionRequest,
 	namespaceID namespace.ID,
 ) (*workflowservice.StartActivityExecutionRequest, error) {
@@ -397,7 +398,7 @@ func (h *frontendHandler) validateAndPopulateStartRequest(
 	}
 
 	if cbs := req.GetCompletionCallbacks(); len(cbs) > 0 {
-		if err := h.callbackValidator.Validate(req.GetNamespace(), cbs); err != nil {
+		if err := h.callbackValidator.Validate(ctx, req.GetNamespace(), cbs); err != nil {
 			return nil, err
 		}
 	}

chasm/lib/activity/frontend_test.go

@@ -1,6 +1,7 @@
 package activity
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -50,11 +51,11 @@ func TestRequestIdStableAcrossRetries(t *testing.T) {
 	// validateAndPopulateStartRequest with the same request pointer.
 	validateTwoAttempts := func(t *testing.T, req *workflowservice.StartActivityExecutionRequest) {
 		t.Helper()
-		clone1, err := h.validateAndPopulateStartRequest(req, nsID)
+		clone1, err := h.validateAndPopulateStartRequest(context.Background(), req, nsID)
 		require.NoError(t, err)
 		require.NotEmpty(t, clone1.RequestId)
 
-		clone2, err := h.validateAndPopulateStartRequest(req, nsID)
+		clone2, err := h.validateAndPopulateStartRequest(context.Background(), req, nsID)
 		require.NoError(t, err)
 		require.Equal(t, clone1.RequestId, clone2.RequestId)
 	}

chasm/lib/callback/validator.go

@@ -1,6 +1,7 @@
 package callback
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
@@ -11,7 +12,11 @@ import (
 )
 
 // Validator validates completion callbacks attached to executions (workflows and standalone activities).
-type Validator struct {
+type Validator interface {
+	Validate(ctx context.Context, namespaceName string, cbs []*commonpb.Callback) error
+}
+
+type validator struct {
 	maxCallbacksPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter
 	urlMaxLength             dynamicconfig.IntPropertyFnWithNamespaceFilter
 	headerMaxSize            dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -23,8 +28,8 @@ func NewValidator(
 	urlMaxLength dynamicconfig.IntPropertyFnWithNamespaceFilter,
 	headerMaxSize dynamicconfig.IntPropertyFnWithNamespaceFilter,
 	endpointRules dynamicconfig.TypedPropertyFnWithNamespaceFilter[AddressMatchRules],
-) *Validator {
-	return &Validator{
+) Validator {
+	return &validator{
 		maxCallbacksPerExecution: maxCallbacksPerExecution,
 		urlMaxLength:             urlMaxLength,
 		headerMaxSize:            headerMaxSize,
@@ -34,7 +39,7 @@ func NewValidator(
 
 // Validate validates completion callbacks: count, URL length, endpoint allowlist, header size, and normalizes header
 // keys to lowercase.
-func (v *Validator) Validate(namespaceName string, cbs []*commonpb.Callback) error {
+func (v *validator) Validate(_ context.Context, namespaceName string, cbs []*commonpb.Callback) error {
 	if len(cbs) > v.maxCallbacksPerExecution(namespaceName) {
 		return serviceerror.NewInvalidArgumentf(
 			"cannot attach more than %d callbacks to an execution", v.maxCallbacksPerExecution(namespaceName),

chasm/lib/callback/validator_test.go

@@ -1,6 +1,7 @@
 package callback
 
 import (
+	"context"
 	"regexp"
 	"testing"
 
@@ -31,7 +32,7 @@ func TestValidateCallbacks(t *testing.T) {
 				},
 			}},
 		}
-		err := v.Validate("ns", cbs)
+		err := v.Validate(context.Background(), "ns", cbs)
 		require.NoError(t, err)
 	})
 
@@ -46,7 +47,7 @@ func TestValidateCallbacks(t *testing.T) {
 			{Variant: &commonpb.Callback_Nexus_{Nexus: &commonpb.Callback_Nexus{Url: "http://localhost/cb1"}}},
 			{Variant: &commonpb.Callback_Nexus_{Nexus: &commonpb.Callback_Nexus{Url: "http://localhost/cb2"}}},
 		}
-		err := v.Validate("ns", cbs)
+		err := v.Validate(context.Background(), "ns", cbs)
 		var invalidArgErr *serviceerror.InvalidArgument
 		require.ErrorAs(t, err, &invalidArgErr)
 		require.Contains(t, err.Error(), "cannot attach more than 1 callbacks")
@@ -66,7 +67,7 @@ func TestValidateCallbacks(t *testing.T) {
 				},
 			}},
 		}
-		err := v.Validate("ns", cbs)
+		err := v.Validate(context.Background(), "ns", cbs)
 		var invalidArgErr *serviceerror.InvalidArgument
 		require.ErrorAs(t, err, &invalidArgErr)
 		require.Contains(t, err.Error(), "url length longer than max length allowed")
@@ -81,7 +82,7 @@ func TestValidateCallbacks(t *testing.T) {
 				},
 			}},
 		}
-		err := v.Validate("ns", cbs)
+		err := v.Validate(context.Background(), "ns", cbs)
 		var invalidArgErr *serviceerror.InvalidArgument
 		require.ErrorAs(t, err, &invalidArgErr)
 		require.Contains(t, err.Error(), "header size longer than max allowed size")
@@ -96,7 +97,7 @@ func TestValidateCallbacks(t *testing.T) {
 				},
 			}},
 		}
-		err := v.Validate("ns", cbs)
+		err := v.Validate(context.Background(), "ns", cbs)
 		require.NoError(t, err)
 		nexus := cbs[0].GetNexus()
 		require.Equal(t, "application/json", nexus.Header["content-type"])
@@ -119,7 +120,7 @@ func TestValidateCallbacks(t *testing.T) {
 				},
 			}},
 		}
-		err := v.Validate("ns", cbs)
+		err := v.Validate(context.Background(), "ns", cbs)
 		var invalidArgErr *serviceerror.InvalidArgument
 		require.ErrorAs(t, err, &invalidArgErr)
 		require.Contains(t, err.Error(), "does not match any configured callback address")
@@ -129,14 +130,14 @@ func TestValidateCallbacks(t *testing.T) {
 		cbs := []*commonpb.Callback{
 			{Variant: nil},
 		}
-		err := v.Validate("ns", cbs)
+		err := v.Validate(context.Background(), "ns", cbs)
 		var unimplementedErr *serviceerror.Unimplemented
 		require.ErrorAs(t, err, &unimplementedErr)
 		require.Contains(t, err.Error(), "unknown callback variant")
 	})
 
 	t.Run("EmptyCallbacksNoError", func(t *testing.T) {
-		err := v.Validate("ns", nil)
+		err := v.Validate(context.Background(), "ns", nil)
 		require.NoError(t, err)
 	})
 
@@ -146,7 +147,7 @@ func TestValidateCallbacks(t *testing.T) {
 				Internal: &commonpb.Callback_Internal{},
 			}},
 		}
-		err := v.Validate("ns", cbs)
+		err := v.Validate(context.Background(), "ns", cbs)
 		require.NoError(t, err)
 	})
 }

service/frontend/fx.go

@@ -827,7 +827,7 @@ func OperatorHandlerProvider(
 // so that existing operator configurations (component.callbacks.allowedAddresses) are honored.
 // TODO: Once HSM callbacks (components/callbacks) are removed, move this provider into
 // chasm/lib/callback/fx.go and read directly from callback.AllowedAddresses.
-func callbackValidatorProvider(dc *dynamicconfig.Collection) *callback.Validator {
+func callbackValidatorProvider(dc *dynamicconfig.Collection) callback.Validator {
 	return callback.NewValidator(
 		callback.MaxPerExecution.Get(dc),
 		dynamicconfig.FrontendCallbackURLMaxLength.Get(dc),
@@ -876,7 +876,7 @@ func HandlerProvider(
 	healthInterceptor *interceptor.HealthInterceptor,
 	scheduleSpecBuilder *scheduler.SpecBuilder,
 	activityHandler activity.FrontendHandler,
-	callbackValidator *callback.Validator,
+	callbackValidator callback.Validator,
 	nexusOperationHandler chasmnexus.FrontendHandler,
 	registry *chasm.Registry,
 	frontendServiceResolver membership.ServiceResolver,

service/frontend/workflow_handler.go

@@ -123,7 +123,7 @@ type (
 
 		status int32
 
-		callbackValidator               *callback.Validator
+		callbackValidator               callback.Validator
 		tokenSerializer                 *tasktoken.Serializer
 		config                          *Config
 		versionChecker                  headers.VersionChecker
@@ -302,7 +302,7 @@ func (wh *WorkflowHandler) ValidateWorkerDeploymentVersionComputeConfig(
 
 // NewWorkflowHandler creates a gRPC handler for workflowservice
 func NewWorkflowHandler(
-	callbackValidator *callback.Validator,
+	callbackValidator callback.Validator,
 	config *Config,
 	namespaceReplicationQueue persistence.NamespaceReplicationQueue,
 	visibilityMgr manager.VisibilityManager,
@@ -545,7 +545,7 @@ func (wh *WorkflowHandler) StartWorkflowExecution(
 	defer log.CapturePanic(wh.logger, &retError)
 
 	var err error
-	if request, err = wh.prepareStartWorkflowRequest(request); err != nil {
+	if request, err = wh.prepareStartWorkflowRequest(ctx, request); err != nil {
 		return nil, err
 	}
 
@@ -603,6 +603,7 @@ func (wh *WorkflowHandler) convertToStartWorkflowExecutionResponse(
 
 // Validates the request and sets default values where they are missing.
 func (wh *WorkflowHandler) prepareStartWorkflowRequest(
+	ctx context.Context,
 	request *workflowservice.StartWorkflowExecutionRequest,
 ) (*workflowservice.StartWorkflowExecutionRequest, error) {
 	if request == nil {
@@ -680,7 +681,7 @@ func (wh *WorkflowHandler) prepareStartWorkflowRequest(
 	}
 
 	if cbs := request.GetCompletionCallbacks(); len(cbs) > 0 {
-		if err := wh.callbackValidator.Validate(namespaceName.String(), cbs); err != nil {
+		if err := wh.callbackValidator.Validate(ctx, namespaceName.String(), cbs); err != nil {
 			return nil, err
 		}
 	}
@@ -791,7 +792,7 @@ func (wh *WorkflowHandler) ExecuteMultiOperation(
 		return nil, errMultiOpNotStartAndUpdate
 	}
 
-	historyReq, err := wh.convertToHistoryMultiOperationRequest(namespaceID, request)
+	historyReq, err := wh.convertToHistoryMultiOperationRequest(ctx, namespaceID, request)
 	if err != nil {
 		return nil, err
 	}
@@ -815,6 +816,7 @@ func (wh *WorkflowHandler) ExecuteMultiOperation(
 }
 
 func (wh *WorkflowHandler) convertToHistoryMultiOperationRequest(
+	ctx context.Context,
 	namespaceID namespace.ID,
 	request *workflowservice.ExecuteMultiOperationRequest,
 ) (*historyservice.ExecuteMultiOperationRequest, error) {
@@ -825,7 +827,7 @@ func (wh *WorkflowHandler) convertToHistoryMultiOperationRequest(
 	errs := make([]error, len(request.Operations))
 
 	for i, op := range request.Operations {
-		convertedOp, opWorkflowID, err := wh.convertToHistoryMultiOperationItem(namespaceID, namespace.Name(request.Namespace), op)
+		convertedOp, opWorkflowID, err := wh.convertToHistoryMultiOperationItem(ctx, namespaceID, namespace.Name(request.Namespace), op)
 		if err != nil {
 			hasError = true
 		} else {
@@ -856,6 +858,7 @@ func (wh *WorkflowHandler) convertToHistoryMultiOperationRequest(
 }
 
 func (wh *WorkflowHandler) convertToHistoryMultiOperationItem(
+	ctx context.Context,
 	namespaceID namespace.ID,
 	namespaceName namespace.Name,
 	op *workflowservice.ExecuteMultiOperationRequest_Operation,
@@ -868,7 +871,7 @@ func (wh *WorkflowHandler) convertToHistoryMultiOperationItem(
 			return nil, "", errMultiOpNamespaceMismatch
 		}
 		var err error
-		if startReq, err = wh.prepareStartWorkflowRequest(startReq); err != nil {
+		if startReq, err = wh.prepareStartWorkflowRequest(ctx, startReq); err != nil {
 			return nil, "", err
 		}
 		if len(startReq.CronSchedule) > 0 {