Add validation for a few string fields (#5487)

Add string validation for a few string fields

Since we disable utf8 string validation from proto level, we want to
enforce minimal validation for some key fields.

Unit tests

No

<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->

<!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) -->

Author: yiminc

common/searchattribute/encode_value.go

@@ -25,15 +25,19 @@
 package searchattribute
 
 import (
+	"errors"
 	"fmt"
 	"time"
+	"unicode/utf8"
 
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 
 	"go.temporal.io/server/common/payload"
 )
 
+var ErrInvalidString = errors.New("SearchAttribute value is not a valid UTF-8 string")
+
 // EncodeValue encodes search attribute value and IndexedValueType to Payload.
 func EncodeValue(val interface{}, t enumspb.IndexedValueType) (*commonpb.Payload, error) {
 	valPayload, err := payload.Encode(val)
@@ -70,16 +74,37 @@ func DecodeValue(
 	case enumspb.INDEXED_VALUE_TYPE_INT:
 		return decodeValueTyped[int64](value, allowList)
 	case enumspb.INDEXED_VALUE_TYPE_KEYWORD:
-		return decodeValueTyped[string](value, allowList)
+		return validateStrings(decodeValueTyped[string](value, allowList))
 	case enumspb.INDEXED_VALUE_TYPE_TEXT:
-		return decodeValueTyped[string](value, allowList)
+		return validateStrings(decodeValueTyped[string](value, allowList))
 	case enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST:
-		return decodeValueTyped[[]string](value, false)
+		return validateStrings(decodeValueTyped[[]string](value, false))
 	default:
 		return nil, fmt.Errorf("%w: %v", ErrInvalidType, t)
 	}
 }
 
+func validateStrings(anyValue any, err error) (any, error) {
+	if err != nil {
+		return anyValue, err
+	}
+
+	// validate strings
+	switch value := anyValue.(type) {
+	case string:
+		if !utf8.ValidString(value) {
+			return nil, fmt.Errorf("%w: %s", ErrInvalidString, value)
+		}
+	case []string:
+		for _, item := range value {
+			if !utf8.ValidString(item) {
+				return nil, fmt.Errorf("%w: %s", ErrInvalidString, item)
+			}
+		}
+	}
+	return anyValue, err
+}
+
 // decodeValueTyped tries to decode to the given type.
 // If the input is a list and allowList is false, then it will return only the first element.
 // If the input is a list and allowList is true, then it will return the decoded list.

common/searchattribute/encode_value_test.go

@@ -25,6 +25,7 @@
 package searchattribute
 
 import (
+	"errors"
 	"testing"
 	"time"
 
@@ -372,3 +373,21 @@ func Test_EncodeValue(t *testing.T) {
 		"Datetime Search Attribute is expected to be encoded in RFC 3339 format")
 	s.Equal("Datetime", string(encodedPayload.Metadata["type"]))
 }
+
+func Test_ValidateStrings(t *testing.T) {
+	_, err := validateStrings("anything here", errors.New("test error"))
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "test error")
+
+	_, err = validateStrings("\x87\x01", nil)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "is not a valid UTF-8 string")
+
+	value, err := validateStrings("anything here", nil)
+	assert.Nil(t, err)
+	assert.Equal(t, "anything here", value)
+
+	_, err = validateStrings([]string{"abc", "\x87\x01"}, nil)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "is not a valid UTF-8 string")
+}

common/util.go

@@ -32,6 +32,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"unicode/utf8"
 
 	"github.com/dgryski/go-farm"
 	"github.com/gogo/protobuf/proto"
@@ -720,3 +721,10 @@ func OverrideWorkflowTaskTimeout(
 func CloneProto[T proto.Message](v T) T {
 	return proto.Clone(v).(T)
 }
+
+func ValidateUTF8String(fieldName string, strValue string) error {
+	if !utf8.ValidString(strValue) {
+		return serviceerror.NewInvalidArgument(fmt.Sprintf("%s %v is not a valid UTF-8 string", fieldName, strValue))
+	}
+	return nil
+}

service/frontend/workflow_handler.go

@@ -380,6 +380,9 @@ func (wh *WorkflowHandler) StartWorkflowExecution(ctx context.Context, request *
 		return nil, errWorkflowTypeTooLong
 	}
 
+	if err := common.ValidateUTF8String("WorkflowType", request.WorkflowType.GetName()); err != nil {
+		return nil, err
+	}
 	if err := wh.validateTaskQueue(request.TaskQueue); err != nil {
 		return nil, err
 	}
@@ -388,12 +391,12 @@ func (wh *WorkflowHandler) StartWorkflowExecution(ctx context.Context, request *
 		return nil, err
 	}
 
-	if request.GetRequestId() == "" {
+	if request.RequestId == "" {
 		return nil, errRequestIDNotSet
 	}
 
-	if len(request.GetRequestId()) > wh.config.MaxIDLengthLimit() {
-		return nil, errRequestIDTooLong
+	if err := validateRequestId(&request.RequestId, wh.config.MaxIDLengthLimit()); err != nil {
+		return nil, err
 	}
 
 	request, err := wh.unaliasStartWorkflowExecutionRequestSearchAttributes(request, namespaceName)
@@ -2002,12 +2005,16 @@ func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(ctx context.Context,
 		return nil, errWorkflowTypeTooLong
 	}
 
+	if err := common.ValidateUTF8String("WorkflowType", request.WorkflowType.GetName()); err != nil {
+		return nil, err
+	}
+
 	if err := wh.validateTaskQueue(request.TaskQueue); err != nil {
 		return nil, err
 	}
 
-	if len(request.GetRequestId()) > wh.config.MaxIDLengthLimit() {
-		return nil, errRequestIDTooLong
+	if err := validateRequestId(&request.RequestId, wh.config.MaxIDLengthLimit()); err != nil {
+		return nil, err
 	}
 
 	if err := wh.validateSignalWithStartWorkflowTimeouts(request); err != nil {
@@ -4348,6 +4355,9 @@ func (wh *WorkflowHandler) validateTaskQueue(t *taskqueuepb.TaskQueue) error {
 	if len(t.GetName()) > wh.config.MaxIDLengthLimit() {
 		return errTaskQueueTooLong
 	}
+	if err := common.ValidateUTF8String("TaskQueue", t.GetName()); err != nil {
+		return err
+	}
 
 	enums.SetDefaultTaskQueueKind(&t.Kind)
 	return nil
@@ -4356,26 +4366,33 @@ func (wh *WorkflowHandler) validateTaskQueue(t *taskqueuepb.TaskQueue) error {
 func (wh *WorkflowHandler) validateBuildIdOrderingUpdate(
 	req *workflowservice.UpdateWorkerBuildIdOrderingRequest,
 ) error {
-	errstr := "request to update worker build id ordering requires:"
-	hadErr := false
+	errDeets := []string{"request to update worker build id compatability requires: "}
+
+	checkIdLen := func(id string) {
+		if len(id) > wh.config.WorkerBuildIdSizeLimit() {
+			errDeets = append(errDeets, fmt.Sprintf(" Worker build IDs to be no larger than %v characters",
+				wh.config.WorkerBuildIdSizeLimit()))
+		}
+
+		if err := common.ValidateUTF8String("BuildId", id); err != nil {
+			errDeets = append(errDeets, err.Error())
+		}
+	}
+
 	if req.GetNamespace() == "" {
-		errstr += " `namespace` to be set"
-		hadErr = true
+		errDeets = append(errDeets, "`namespace` to be set")
 	}
 	if req.GetTaskQueue() == "" {
-		errstr += " `task_queue` to be set"
-		hadErr = true
+		errDeets = append(errDeets, "`task_queue` to be set")
 	}
 	if req.GetVersionId().GetWorkerBuildId() == "" {
-		errstr += " targeting a valid version identifier"
-		hadErr = true
-	}
-	if len(req.GetVersionId().GetWorkerBuildId()) > wh.config.WorkerBuildIdSizeLimit() {
-		errstr += fmt.Sprintf(" Worker build IDs to be no larger than %v characters", wh.config.WorkerBuildIdSizeLimit())
-		hadErr = true
+		errDeets = append(errDeets, "targeting a valid version identifier")
+	} else {
+		checkIdLen(req.GetVersionId().GetWorkerBuildId())
 	}
-	if hadErr {
-		return serviceerror.NewInvalidArgument(errstr)
+
+	if len(errDeets) > 1 {
+		return serviceerror.NewInvalidArgument(strings.Join(errDeets, ", "))
 	}
 	return nil
 }
@@ -4657,6 +4674,24 @@ func (wh *WorkflowHandler) validateRetryPolicy(namespaceName namespace.Name, ret
 	return common.ValidateRetryPolicy(retryPolicy)
 }
 
+func validateRequestId(requestID *string, lenLimit int) error {
+	if requestID == nil {
+		// should never happen, but just in case.
+		return serviceerror.NewInvalidArgument("RequestId is nil")
+	}
+	if *requestID == "" {
+		// For easy direct API use, we default the request ID here but expect all
+		// SDKs and other auto-retrying clients to set it
+		*requestID = uuid.New()
+	}
+
+	if len(*requestID) > lenLimit {
+		return errRequestIDTooLong
+	}
+
+	return common.ValidateUTF8String("RequestId", *requestID)
+}
+
 func (wh *WorkflowHandler) validateStartWorkflowTimeouts(
 	request *workflowservice.StartWorkflowExecutionRequest,
 ) error {
@@ -4753,7 +4788,7 @@ func (wh *WorkflowHandler) makeFakeContinuedAsNewEvent(
 func (wh *WorkflowHandler) validateNamespace(
 	namespace string,
 ) error {
-	if err := wh.validateUTF8String(namespace); err != nil {
+	if err := common.ValidateUTF8String("Namespace", namespace); err != nil {
 		return err
 	}
 	if len(namespace) > wh.config.MaxIDLengthLimit() {
@@ -4768,7 +4803,7 @@ func (wh *WorkflowHandler) validateWorkflowID(
 	if workflowID == "" {
 		return errWorkflowIDNotSet
 	}
-	if err := wh.validateUTF8String(workflowID); err != nil {
+	if err := common.ValidateUTF8String("WorkflowId", workflowID); err != nil {
 		return err
 	}
 	if len(workflowID) > wh.config.MaxIDLengthLimit() {

service/frontend/workflow_handler_test.go

@@ -2655,6 +2655,18 @@ func TestContextNearDeadline(t *testing.T) {
 	assert.False(t, contextNearDeadline(ctx, time.Millisecond))
 }
 
+func TestValidateRequestId(t *testing.T) {
+	req := workflowservice.StartWorkflowExecutionRequest{RequestId: ""}
+	err := validateRequestId(&req.RequestId, 100)
+	assert.Nil(t, err)
+	assert.Len(t, req.RequestId, 36) // new UUID length
+
+	req.RequestId = "\x87\x01"
+	err = validateRequestId(&req.RequestId, 100)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "not a valid UTF-8 string")
+}
+
 func (s *workflowHandlerSuite) Test_DeleteWorkflowExecution() {
 	config := s.newConfig()
 	wh := s.getWorkflowHandler(config)

service/history/commandChecker.go

@@ -380,6 +380,9 @@ func (v *commandAttrValidator) validateTimerScheduleAttributes(
 	if len(attributes.GetTimerId()) > v.maxIDLengthLimit {
 		return failedCause, serviceerror.NewInvalidArgument("TimerId exceeds length limit.")
 	}
+	if err := common.ValidateUTF8String("TimerId", attributes.TimerId); err != nil {
+		return failedCause, err
+	}
 	if timestamp.DurationValue(attributes.GetStartToFireTimeout()) <= 0 {
 		return failedCause, serviceerror.NewInvalidArgument("A valid StartToFireTimeout is not set on command.")
 	}
@@ -498,7 +501,22 @@ func (v *commandAttrValidator) validateCancelExternalWorkflowExecutionAttributes
 	if len(attributes.GetWorkflowId()) > v.maxIDLengthLimit {
 		return failedCause, serviceerror.NewInvalidArgument("WorkflowId exceeds length limit.")
 	}
-	runID := attributes.GetRunId()
+	runID := attributes.RunId
+	workflowID := attributes.WorkflowId
+	ns := attributes.Namespace
+
+	if workflowID == "" {
+		return failedCause, serviceerror.NewInvalidArgument(fmt.Sprintf("WorkflowId is not set on RequestCancelExternalWorkflowExecutionCommand. Namespace=%s RunId=%s", ns, runID))
+	}
+	if len(ns) > v.maxIDLengthLimit {
+		return failedCause, serviceerror.NewInvalidArgument(fmt.Sprintf("Namespace on RequestCancelExternalWorkflowExecutionCommand exceeds length limit. WorkflowId=%s RunId=%s Namespace=%s Length=%d Limit=%d", workflowID, runID, ns, len(ns), v.maxIDLengthLimit))
+	}
+	if len(workflowID) > v.maxIDLengthLimit {
+		return failedCause, serviceerror.NewInvalidArgument(fmt.Sprintf("WorkflowId on RequestCancelExternalWorkflowExecutionCommand exceeds length limit. WorkflowId=%s Length=%d Limit=%d RunId=%s Namespace=%s", workflowID, len(workflowID), v.maxIDLengthLimit, runID, ns))
+	}
+	if err := common.ValidateUTF8String("WorkflowId", workflowID); err != nil {
+		return failedCause, err
+	}
 	if runID != "" && uuid.Parse(runID) == nil {
 		return failedCause, serviceerror.NewInvalidArgument("Invalid RunId set on command.")
 	}
@@ -540,6 +558,22 @@ func (v *commandAttrValidator) validateSignalExternalWorkflowExecutionAttributes
 	}
 
 	targetRunID := attributes.Execution.GetRunId()
+	signalName := attributes.SignalName
+	workflowID := attributes.Execution.WorkflowId
+	ns := attributes.Namespace
+
+	if workflowID == "" {
+		return failedCause, serviceerror.NewInvalidArgument(fmt.Sprintf("WorkflowId is not set on SignalExternalWorkflowExecutionCommand. Namespace=%s RunId=%s SignalName=%s", ns, targetRunID, signalName))
+	}
+	if len(ns) > v.maxIDLengthLimit {
+		return failedCause, serviceerror.NewInvalidArgument(fmt.Sprintf("Namespace on SignalExternalWorkflowExecutionCommand exceeds length limit. WorkflowId=%s Namespace=%s Length=%d Limit=%d RunId=%s SignalName=%s", workflowID, ns, len(ns), v.maxIDLengthLimit, targetRunID, signalName))
+	}
+	if len(workflowID) > v.maxIDLengthLimit {
+		return failedCause, serviceerror.NewInvalidArgument(fmt.Sprintf("WorkflowId on SignalExternalWorkflowExecutionCommand exceeds length limit. WorkflowId=%s Length=%d Limit=%d Namespace=%s RunId=%s SignalName=%s", workflowID, len(workflowID), v.maxIDLengthLimit, ns, targetRunID, signalName))
+	}
+	if err := common.ValidateUTF8String("WorkflowId", workflowID); err != nil {
+		return failedCause, err
+	}
 	if targetRunID != "" && uuid.Parse(targetRunID) == nil {
 		return failedCause, serviceerror.NewInvalidArgument("Invalid RunId set on command.")
 	}
@@ -791,6 +825,14 @@ func (v *commandAttrValidator) validateStartChildExecutionAttributes(
 		return failedCause, serviceerror.NewInvalidArgument("WorkflowType exceeds length limit.")
 	}
 
+	if err := common.ValidateUTF8String("WorkflowId", attributes.WorkflowId); err != nil {
+		return failedCause, err
+	}
+
+	if err := common.ValidateUTF8String("WorkflowType", attributes.WorkflowType.Name); err != nil {
+		return failedCause, err
+	}
+
 	if timestamp.DurationValue(attributes.GetWorkflowExecutionTimeout()) < 0 {
 		return failedCause, serviceerror.NewInvalidArgument("Invalid WorkflowExecutionTimeout.")
 	}
@@ -868,6 +910,10 @@ func (v *commandAttrValidator) validateTaskQueue(
 		return taskQueue, serviceerror.NewInvalidArgument(fmt.Sprintf("task queue name exceeds length limit of %v", v.maxIDLengthLimit))
 	}
 
+	if err := common.ValidateUTF8String("TaskQueue", name); err != nil {
+		return taskQueue, err
+	}
+
 	if strings.HasPrefix(name, reservedTaskQueuePrefix) {
 		return taskQueue, serviceerror.NewInvalidArgument(fmt.Sprintf("task queue name cannot start with reserved prefix %v", reservedTaskQueuePrefix))
 	}